Реализация сервиса на C++: TDD, DDD и событийно-ориентированная архитектура

Привет!

Данная статья посвящена описанию реализации учебного проекта ^[1]. Проект является С++ реализацией сервиса по распределению позиций заказов внутри партий. Исходная реализация данного сервиса представлена на Python в книге «Паттерны разработки на Python: TDD, DDD и событийно-ориентированная архитектура».

Читателю рекомендуется ознакомиться с оригиналом проекта ^[2] и книгой ^[3] «Паттерны разработки на Python: TDD, DDD и событийно-ориентированная архитектура».

Содержание

1. Архитектура приложений ^[4]

2. Гексагональная архитектура ^[5]

3. Доменный слой ^[6]

4. Описание портов ^[7]

5. Точки входа ^[8]

6. Сервисный слой ^[9]

7. Слой адаптеров ^[10]

8. Тестирование. Методология TDD ^[11]

9. Заключение ^[12]

10. Список литературы ^[13]

11. Благодарность ^[14]

Соглашения в проекте

• Окончание T_Ptr в имени типа означает std::shared_ptr<T>.

• Каждый слой приложения вынесен в отдельный namespace:

  • Allocation — глобальный

  • Domain — домен

  • ServiceLayer — сервисный слой

  • Adapters — адаптеры

  • Entrypoints — точки входа

Структура проекта

Используемые термины и директории:

• Первичные адаптеры — точки входа (директория Entrypoints)

• Вторичные адаптеры — адаптеры (директория Adapters)

• Бизнес-логика — домен (директория Domain)

• Сервисный слой — директория ServiceLayer

• утилиты - директория Utilities

Инфраструктура

• PostgreSQL — объектно-реляционная система управления базами данных.

• Redis — нереляционная система управления базами данных класса NoSQL.

Библиотеки и фреймворки

• Poco — библиотека для разработки сетевых приложений.

• Google Test — фреймворк для модульного тестирования.

• pytest — фреймворк для тестирования кода на Python.

Архитектура приложений

Трёхуровневая архитектура

Классическим подходом, описанным Мартином Фаулером, является использование трёхуровневой архитектуры. В ней каждый вышестоящий слой использует нижестоящий. Выделяют три основных слоя:

• Представление — отображение данных, обработка событий пользовательского интерфейса, обслуживание HTTP-запросов, поддержка командной строки и пакетных API-вызовов.

• Домен — бизнес-логика приложения.

• Источник данных — доступ к базе данных, обмен сообщениями, управление транзакциями и т. д.

Пример:

// Слой источника данных
class DataGateway {
public:
    std::string getData() {
        // Имитация получения данных из базы
        return "Данные из базы данных";
    }
};

// Слой домена
class BusinessLogic {
public:
    DataGateway dataGateway;

    std::string processData() {
        std::string data = dataGateway.getData();
        // Простая обработка данных
        return "Обработанные данные: " + data;
    }
};

// Слой представления
class UserInterface {
public:
    BusinessLogic businessLogic;

    void showData() {
        std::string result = businessLogic.processData();
        std::cout << result << std::endl;
    }
};

Такая структура приложения имеет недостатки: каждый слой «сверху» знает о всех «нижележащих», бизнес-логика оказывается связана с конкретным источником данных, а слой представления включает в себя бизнес-логику и транзитивно слой данных. Эти недостатки приводят к связанности, снижают гибкость и усложняют тестирование.

Гексагональная архитектура

Развитием идей трёхслойной архитектуры является гексагональная архитектура (луковичная архитектура), где в центре приложения находится бизнес-логика (домен), выраженная тактическими паттернами DDD.
Домен предоставляет интерфейсы — порты, которые реализуются в остальных частях проекта — адаптерах. Такой подход реализует принцип Dependency Inversion Principle (DIP, SOLID), позволяя, например, подменять настоящий адаптер фиктивным или заменять один адаптер другим.

Адаптеры делятся на два типа:

• Первичные, ведущие адаптеры (driving adapters) — с которых начинается поток выполнения (например, HTTP-запрос, команда из CLI, сообщение из очереди).

• Вторичные, ведомые адаптеры (driven adapters) — которые вызываются из домена (например, работа с БД или отправка сообщений).

Пример потока выполнения:

Сервисный слой

Кроме задач бизнес-логики и адаптеров, в приложении возникают и общие инфраструктурные задачи:

• управление транзакциями во время выполнения бизнес-сценариев;

• единое место сбора и запуска бизнес-сценариев;

• вызов обработчиков по событиям, полученным из доменного слоя;

• координация работы нескольких адаптеров.

Для таких задач вводят дополнительный слой — сервисный слой (Service Layer, Application Layer).

Определение:

«Слой служб (сервисный слой) определяет границы приложения и множество операций, предоставляемых им для интерфейсных клиентских слоёв кода. Он инкапсулирует бизнес-логику приложения, управляет транзакциями и координирует реакции на действия».
— Мартин Фаулер, Шаблоны корпоративных приложений.

Сервисный слой — слой, который координирует выполнение бизнес-сценариев, управляет транзакциями и инфраструктурными аспектами, не относящимися напрямую к предметной области. Он служит связующим звеном между доменом и адаптерами, обеспечивая целостность сценариев и правильный порядок их выполнения.

Важно: сервисный слой должен работать не с конкретными адаптерами, а с их абстракциями (портами). Зависимости внедряются через Dependency Injection.

Доменный слой

Важно: в параграфе рассматривается необходимая часть понятий методологии DDD в контексте проекта.

Определения:

Область (domain) — предметная область, для которой разрабатывается программное обеспечение.

Модель (model) — описывает отдельные аспекты области и может быть использована для решения проблемы.

Единый язык (Ubiquitous Language) — однозначно определённая система понятий используемых стейкхолдерами для описания модели.

Ограниченный контекст (Bounded Context) — чётко определённая граница, внутри которой действует модель.

Описание модели

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

• Артикул (sku) — идентификатор продукции.

• Количество (qty) — объём продукции в партии.

• Ссылка (ref) — уникальный идентификатор партии.

• Ожидаемая дата поставки (eta) — указывается, если партия находится в пути; отсутствует, если продукция уже на складе.

Клиентский заказ — заказ конечного клиента на продукцию организации.
Атрибуты и особенности:

• Идентификатор заказа (orderid) — уникальный идентификатор.

• Может включать позиции с разными наименованиями продукции.

• Состоит из одной или нескольких позиций заказа.

Позиция заказа — элемент клиентского заказа, описывающий потребность в конкретном товаре.
Атрибуты:

• Идентификатор заказа (orderid) — ссылка на заказ, частью которого является позиция.

• Артикул (sku) — идентификатор запрашиваемой продукции.

• Количество (qty) — количество запрашиваемой продукции.

Ограничения и сценарии

Клиенты оформляют заказы, которые идентифицируются ссылкой и состоят из позиций заказа (order lines).
Отдел закупок оформляет партии поставок (batches) продукции.

Задача: разместить позиции заказов в партиях поставки.

Правила размещения

1. После размещения x единиц товара в партия поставки её доступное количество уменьшается на x.
   Пример: партия поставки СТОЛ-МАЛЫЙ, 20 шт.; размещено 2 шт. → остаётся 18 шт.

2. Нельзя разместить позиций заказа, если доступное количество меньше требуемого.
   Пример: партия поставки ПОДУШКА-СИНЯЯ, 1 шт.; позиция заказа — 2 шт. → размещение невозможно.

3. Одну и ту же позиций заказа нельзя разместить дважды в одной партии поставки.
   Пример: партия поставки ВАЗА-СИНЯЯ, 10 шт.; позиция заказа — 2 шт.; если попытаться разместить снова, остаток будет 8 шт., а не 6.

4. Партии имеют предполагаемое время прибытия (eta, estimated arrival time):

   • партия поставки на складе имеют приоритет при размещении;

   • для партии поставки в пути приоритет определяется по eta: чем раньше прибытие, тем выше приоритет.

Паттерны и их реализация

Для реализации описанной модели в DDD применяются паттерны:

• Сущность (Entity)

• Объект-значение (Value Object)

• Агрегат (Aggregate)

• Доменные события и команды

Сущность

Сущность (Entity) — объект, идентифицируемый уникальным свойством. Партии поставки заказа уникальны и определяются по ссылке (ref).

Это проявляется:

• в операторе проверки равенства партий;

• в схеме БД.

Код

namespace Allocation::Domain
{
    /// Представляет партию поставки продукции для распределения.
    class Batch
    {
    public:
        ...

        /// Проверяет, можно ли распределить позицию заказа в партии.
        [[nodiscard]] bool CanAllocate(const OrderLine& line) const noexcept;

        /// Распределяет позицию заказа в партии.
        void Allocate(const OrderLine& line) noexcept;

    private:
        std::string _reference;
        std::string _sku;
        size_t _purchasedQuantity;
        std::optional<std::chrono::year_month_day> _eta;
        std::unordered_set<OrderLine> _allocations;
    };

    bool operator==(const Batch& lhs, const Batch& rhs) noexcept
    {
        return lhs.GetReference() == rhs.GetReference();
    }

    bool operator<(const Batch& lhs, const Batch& rhs) noexcept
    {
        if (!lhs.GetETA().has_value())
            return true;
        if (!rhs.GetETA().has_value())
            return false;
        return lhs.GetETA().value() < rhs.GetETA().value();
    }
}

Схема БД

CREATE TABLE allocation.batches (
    id SERIAL PRIMARY KEY,
    reference VARCHAR(255) UNIQUE NOT NULL,
    sku VARCHAR(255) NOT NULL REFERENCES allocation.products(sku),
    _purchased_quantity INTEGER NOT NULL CHECK (_purchased_quantity > 0), 
    eta DATE
);

Объект-значение

Объект-значение (Value Object) — объект, который не имеет собственной идентичности. Его уникальность определяется набором полей. Два объекта-значения считаются равными, если их значения полей совпадают.

Классический пример — деньги: если два объекта «деньги» имеют одинаковое значение, они равны.

Позиции заказа (Order line) не имеют собственной идентичности и определяется артикулом продукта (sku), количеством (qty) и ссылкой на заказ (orderid).

Код

namespace Allocation::Domain
{
    /// Представляет позицию заказа для распределения.
    struct OrderLine
    {
        /// Ссылка на заказ клиента.
        std::string orderid;
        std::string sku;
        size_t quantity;

        bool operator==(const OrderLine&) const = default;
    };
}

Схема БД

-- Таблица позиций заказов
CREATE TABLE allocation.order_lines (
    id SERIAL PRIMARY KEY,
    sku VARCHAR(255) NOT NULL,
    qty INTEGER NOT NULL CHECK (qty > 0),
    orderid VARCHAR(255) NOT NULL
);

-- Таблица распределений (связь OrderLine -> Batch)
CREATE TABLE allocation.allocations (
    id SERIAL PRIMARY KEY,
    orderline_id INTEGER NOT NULL REFERENCES allocation.order_lines(id),
    batch_id INTEGER NOT NULL REFERENCES allocation.batches(id)
);

Агрегат

Агрегат (Aggregate) — согласованная совокупность связанных сущностей и объектов-значений, объединяемая в единицу консистентности для управления изменениями. Управление доступом к агрегату осуществляется через его корень, который гарантирует соблюдение бизнес-инвариантов и определяет границы возможных операций над агрегатом.

В сервисе агрегатом является Product, который объединяет все партии поставок продукции с общим артикулом. Корнем агрегата выступает сущность Product, через которую осуществляется доступ ко всем изменениям внутри агрегата.

Код

namespace Allocation::Domain
{
    /// Агрегат-продукт, содержит партии поставок с общим артикулом продукции. 
    /// Реализует бизнес-логику распределения позиций заказа в партиях заказа.
    class Product
    {
    public:
        explicit Product(const std::string& sku, const std::vector<Batch>& batches = {},
            size_t versionNumber = 0, bool isNew = true);

        ...

        /// Распределяет позицию заказа в партии заказа агрегата.
        std::optional<std::string> Allocate(const OrderLine& line);

        /// Изменяет количество продукции в партии заказа.
        bool ChangeBatchQuantity(const std::string& ref, size_t qty);

        [[nodiscard]] size_t GetVersion() const noexcept;

        /// Возвращает сообщения, сгенерированные во время выполнения бизнес-логики.
        [[nodiscard]] const std::vector<Domain::IMessagePtr>& Messages() const noexcept;

        void ClearMessages() noexcept;

    private:
        std::string _sku;
        std::unordered_map<std::string, Batch> _referenceByBatches;
        std::unordered_set<std::string> _modifiedBatchRefs;
        std::vector<Domain::IMessagePtr> _messages;
        size_t _versionNumber;
        bool _isModified;
    };
}

Схема БД

-- Таблица агрегатов продуктов
CREATE TABLE allocation.products (
    sku VARCHAR(255) PRIMARY KEY CHECK (sku <> ''),
    version_number BIGINT NOT NULL DEFAULT 0
);

Доменные события и команды

Доменные события (Domain Event) — структуры, описывающие произошедшие в агрегате изменения. Используются для уведомления компонентов системы о произошедших событиях в агрегате.

События обрабатываются подписчиками, описанными в сервисном слое.

Пример:

namespace Allocation::Domain::Events
{
    /// Событие "Распределена позиция заказа".
    struct Allocated final : public AbstractEvent
    {
        Allocated(std::string orderid, std::string sku, size_t qty, std::string batchref)
            : orderid(std::move(orderid)),
              sku(std::move(sku)),
              qty(qty),
              batchref(std::move(batchref)){};

        [[nodiscard]] std::string Name() const override { return "Allocated"; };

        std::string orderid;
        std::string sku;
        size_t qty;
        std::string batchref;
    };
}

Команды — объекты, описывающие намерение изменить состояние домена. Команды формулируют что нужно сделать, но не содержат реализации бизнес-логики.

• Поступают из первичных адаптеров (REST, Redis и др.).

• Обрабатываются сервисным слоем, который делегирует выполнение соответствующему агрегату.

• Не описывают результат, только действие.

Пример:

namespace Allocation::Domain::Commands
{
    /// Команда "Распределить позицию заказа".
    struct Allocate final : public AbstractCommand
    {
        Allocate(std::string orderid, std::string sku, size_t qty)
            : orderid(std::move(orderid)), sku(std::move(sku)), qty(qty)
        {
        }

        [[nodiscard]] std::string Name() const override { return "Allocate"; };

        std::string orderid;
        std::string sku;
        size_t qty;
    };
}

Сообщение - обобщающее понятие команд и событий. Агрегат возвращает сообщения Messages(), метод ClearMessages() очищает сообщения в агрегате.

Описание портов

Абстракции и интерфейсы определяют контракты для всех реализаций, что соответствует принципу подстановки Барбары Лисков (LSP, SOLID). В гексагональной архитектуре контракты формируют порты — интерфейсы, определяющие взаимодействие между доменом и внешними компонентами.

Сообщение

Сообщение - обобщающее понятие команд и событий. Домен предоставляет следующий порт:

namespace Allocation::Domain
{
    struct IMessage
    {
        /// Типы сообщений.
        enum class Type : int
        {
            Event,
            Command
        };

        virtual ~IMessage() = default;

        /// Возвращает имя сообщения.
        [[nodiscard]] virtual std::string Name() const = 0;

        /// Возвращает тип сообщения.
        [[nodiscard]] virtual Type GetType() const = 0;
    };
}

Репозиторий

Репозиторий (Repository) — абстракция над системой хранения данных. Он отвечает за сохранение, загрузку и обновление агрегатов, полученных из конкретного хранилища.

Порт для работы с агрегатом Product:

namespace Allocation::Domain
{
    /// Интерфейс репозитория для работы с агрегатами-продуктами в хранилище.
    class IRepository
    {
    public:
        IRepository() = default;
        virtual ~IRepository() = default;

        /// Общий интерфейс для добавления или обновления агрегат-продукта.
        virtual void Add(Domain::ProductPtr product) = 0;

        /// Возвращает агрегат-продукт по его артикулу.
        [[nodiscard]] virtual Domain::ProductPtr Get(const std::string& sku) = 0;

        /// Возвращает агрегат-продукт по идентификатору партии включённого в него.
        [[nodiscard]] virtual Domain::ProductPtr GetByBatchRef(const std::string& batchRef) = 0;

        ...
    };
}

Пример использования:

void Example(IRepository& repo)
{
    auto product = repo.Get("Amazing-table"); // Загружаем агрегат по артикулу
    if (!product)
        product = std::make_shared<Domain::Product>("Amazing-table"); // Или создаём новый

    Domain::Batch batch("b-add", "Amazing-table", 100);
    product->AddBatch(batch);

    repo.Add(product); // Сохраняем изменения
}

В оригинале проекта метод Add используется и для добавления, и для обновления.
В C++ это можно рассматривать как нарушение принципа SRP.
Для разделения обязанностей вводится расширение — IUpdatableRepository:

namespace Allocation::Domain
{
    /// Расширенный интерфейс репозитория для обновления агрегатов.
    class IUpdatableRepository : public IRepository
    {
    public:
        /// Обновляет агрегат-продукт в репозитории.
        virtual void Update(ProductPtr product, size_t oldVersion) = 0;

        ...
    };
}

Применение интерфейса IUpdatableRepository описано далее в сервисном слое.

Единица работы

Единица работы (Unit of Work, UoW) — паттерн, который обеспечивает атомарность выполнения операций. Все изменения в рамках UoW должны быть либо зафиксированы целиком, либо отменены.

При работе с СУБД UoW управляет транзакцией: при ошибке на любом этапе бизнес-процесса или при отсутствии фиксации выполняется откат.

/// Интерфейс единицы работы.
class IUnitOfWork
{
public:
    ...

    /// Подтверждает изменения.
    virtual void Commit() = 0;

    /// Откатывает изменения.
    virtual void RollBack() = 0;

    /// Проверяет, были ли изменения зафиксированы.
    [[nodiscard]] virtual bool IsCommited() const noexcept = 0;

    ...

    /// Возвращает репозиторий хранилища продуктов.
    [[nodiscard]] virtual IRepository& GetProductRepository() = 0;

    /// Возвращает новые сообщения из обработанных агрегатов.
    [[nodiscard]] virtual std::vector<IMessagePtr> GetNewMessages() noexcept = 0;
};

Пример 1 (успешный сценарий):

void Example(IUnitOfWork& uow)
{
    auto& repo = uow.GetProductRepository();
    auto product = repo.Get("Amazing-table");

    if (!product)
        product = std::make_shared<Domain::Product>("Amazing-table");

    Domain::Batch batch("b-add", "Amazing-table", 100);
    product->AddBatch(batch);

    repo.Add(product);
    uow.Commit(); // фиксируем изменения
}

Пример 2 (с откатом):

void Example()
{
    try
    {
        IUnitOfWork& uow = Allocation::SomeUoW(); // конкретная реализация UoW
        auto& repo = uow.GetProductRepository();
        auto product = repo.Get("Amazing-table");

        if (!product)
            product = std::make_shared<Domain::Product>("Amazing-table");

        Domain::Batch batch("b-add", "Amazing-table", 100);
        product->AddBatch(batch);

        repo.Add(product);
        throw std::runtime_error("Boom");
        uow.Commit(); // не будет вызвано
    }
    catch(...)
    {
        // изменения будут отменены автоматически
    }
}

Точки входа

Типы первичных адаптеров

В приложении используется два типа первичных адаптеров:

1. REST-адаптер — обрабатывают входящие HTTP-запросы.

2. Redis-адаптер — принимают сообщения из Redis-каналов.

REST-адаптер

REST-адаптер обрабатывает входящие HTTP-запросы. После маршрутизации фабрика (HandlerFactory) вызывает соответствующий обработчик. Обработчик разбирает данные и запускает бизнес-сценарий через сервисный слой.

Пример обработчика:

namespace Allocation::Entrypoints::Rest::Handlers
{
    void AllocateHandler::handleRequest(
        Poco::Net::HTTPServerRequest& request, Poco::Net::HTTPServerResponse& response)
    {
        response.set("Access-Control-Allow-Origin", "*");

        std::istream& bodyStream = request.stream();
        std::ostringstream body;
        body << bodyStream.rdbuf();
        Poco::JSON::Parser parser;
        auto result = parser.parse(body.str());
        auto json = result.extract<Poco::JSON::Object::Ptr>();

        try
        {
            auto command = Domain::FromJson<Domain::Commands::Allocate>(json);
            ServiceLayer::MessageBus::Instance().Handle(command);

            response.setStatus(Poco::Net::HTTPResponse::HTTP_ACCEPTED);
            response.setContentType("application/json");
            response.send();
            return;
        }
        catch (const Poco::Exception& ex)
        {
            response.setStatus(Poco::Net::HTTPResponse::HTTP_INTERNAL_SERVER_ERROR);
            response.setContentType("application/json");
            std::string msg = ex.displayText();
            response.send() << "{"message":"" << msg << ""}";

            Allocation::Loggers::GetLogger()->Error(msg);
        }
        catch (const std::runtime_error& ex)
        {
            response.setStatus(Poco::Net::HTTPResponse::HTTP_CONFLICT);
            response.setContentType("application/json");
            std::ostream& ostr = response.send();
            std::string msg = ex.what();
            ostr << "{"message": "" << msg << ""}";

            Allocation::Loggers::GetLogger()->Error(msg);
        }
        // ... обработка исключений
    }
}

Принятое соглашение:

• инфраструктурные ошибки (например, проблемы с БД) передаются через Poco::Exception;

• ошибки бизнес-логики — через std::exception и его наследников.

Redis-адаптер

В проекте Redis используется в качестве брокера сообщений и служит каналом для публикации событий между сервисами.
В сервисе реализован компонент RedisListener, который позволяет подписывать обработчики на нужные каналы.

namespace Allocation::Entrypoints::Redis
{
    /// Концепт для обработчика Redis-сообщений.
    template <typename Handler>
    concept RedisMessageHandler = requires(Handler h, const std::string& payload) {
        { h(payload) } -> std::same_as<void>;
    };

    /// Слушает сообщения из Redis и перенаправляет их в обработчики.
    class RedisListener
    {
    public:
        RedisListener()
            : _connection(Adapters::Redis::RedisConnectionPool::Instance().GetConnection()),
              _reader(*static_cast<Poco::Redis::Client::Ptr>(_connection))
        {
            _reader.redisResponse += Poco::delegate(this, &RedisListener::OnRedisMessage);
        }

        ...

        /// Запускает асинхронное чтение сообщений из Redis.
        void Start() { _reader.start(); }

        /// Останавливает асинхронное чтение сообщений из Redis.
        void Stop() { _reader.stop(); };

        /// Подписывается на канал Redis и регистрирует обработчик сообщений.
        template <RedisMessageHandler Handler>
        void Subscribe(const std::string& channel, Handler&& handler)
        {
            Poco::Redis::Array subscribe;
            subscribe.add("SUBSCRIBE").add(channel);

            static_cast<Poco::Redis::Client::Ptr>(_connection)->execute<void>(subscribe);
            static_cast<Poco::Redis::Client::Ptr>(_connection)->flush();
            _handlers.try_emplace(channel, std::forward<Handler>(handler));
        }

    private:
        /// Обрабатывает входящие сообщения от Redis и вызывает обработчики.
        void OnRedisMessage(const void* sender, Poco::Redis::RedisEventArgs& args)
        {
            if (const Poco::Exception* exception = args.exception(); exception)
            {
                Allocation::Loggers::GetLogger()->Error(
                    "Redis exception: " + exception->displayText());
                return;
            }

            try
            {
                if (auto msg = args.message(); msg && msg->isArray())
                {
                    Poco::Redis::Type<Poco::Redis::Array>* arrayType =
                        dynamic_cast<Poco::Redis::Type<Poco::Redis::Array>*>(args.message().get());
                    if (!arrayType)
                        return;
                    Poco::Redis::Array& array = arrayType->value();
                    if (array.size() == 3)
                    {
                        Poco::Redis::BulkString type = array.get<Poco::Redis::BulkString>(0);
                        if (type != "message")
                            return;
                        auto channel = std::string(array.get<Poco::Redis::BulkString>(1));
                        auto payload = std::string(array.get<Poco::Redis::BulkString>(2));

                        if (auto it = _handlers.find(channel); it != _handlers.end())
                            it->second(payload);
                    }
                }
            }
            catch (const Poco::Exception& e)
            {
                Allocation::Loggers::GetLogger()->Error(
                    "RedisListener exception: " + std::string(e.displayText()));
            }
            ...
        }
        ...
    }
};

Пример подписки:

void HandleChangeBatchQuantity(const std::string& payload)
{
    if (payload.empty())
        return;

    Poco::JSON::Parser parser;
    auto parsed = parser.parse(payload);
    auto json = parsed.extract<Poco::JSON::Object::Ptr>();

    auto command = Domain::FromJson<Domain::Commands::ChangeBatchQuantity>(json);
    ServiceLayer::MessageBus::Instance().Handle(command);
}

void Example()
{
    Entrypoints::Redis::RedisListener redisListener;
    redisListener.Subscribe("change_batch_quantity", HandleChangeBatchQuantity);
    redisListener.Start(); // прослушивает в отдельном потоке
}

Сервисный слой

Для цельности повествования необходимо рассмотреть вспомогательные классы — TrackingRepository и AbstractUnitOfWork.

Описание вспомогательных классов

TrackingRepository

TrackingRepository это обёртка (Decorator / Proxy) над репозиторием. Его задача отслеживать агрегаты с которыми работали в контексте репозитория и разделять вызов метода IRepository::Add(Domain::ProductPtr product) на обновление и добавление нового агрегата.

namespace Allocation::Adapters::Repository
{
    /// Репозиторий для отслеживания изменений агрегатов продуктов.
    class TrackingRepository final : public Domain::IUpdatableRepository
    {
        public:
        /// repo - Отслеживаемый репозиторий.
        TrackingRepository(Domain::IUpdatableRepository& repo);

        /// Общий интерфейс для добавления или обновления агрегат-продукта.
        void TrackingRepository::Add(Domain::ProductPtr product)
        {
            if (!product)
                throw std::invalid_argument("The nullptr product");

            auto sku = product->GetSKU();
            if (auto it = _skuToProductAndOldVersion.find(sku); it != _skuToProductAndOldVersion.end())
                Update(product, it->second.second);
            else
            {
                _repo.Add(product);
                product->SetModified(false);
                _skuToProductAndOldVersion.insert({sku, {product, product->GetVersion()}});
            }
        }

        ...

        /// Возвращает отслеживаемые агрегаты.
        [[nodiscard]] std::vector<std::pair<Domain::ProductPtr, size_t>> GetSeen() const noexcept;

        /// Очищает наблюдаемые агрегаты.
        void Clear() noexcept;

        /// Обновляет агрегат-продукт в репозитории.
        void Update(Domain::ProductPtr product, int oldVersion) override
        {
            if (!product)
                throw std::invalid_argument("The nullptr product");

            _repo.Update(product, oldVersion);
            product->SetModified(false);
        }

        private:
            Domain::IUpdatableRepository& _repo;
            std::unordered_map<std::string, std::pair<Domain::ProductPtr, size_t>>
                _skuToProductAndOldVersion;
    };
}

AbstractUnitOfWork

AbstractUnitOfWork отвечает за централизованный сбор сообщений и обновление модифицированных агрегатов в контексте текущего UoW при вызове метода Commit().

namespace Allocation::ServiceLayer::UoW
{
    /// Абстрактный базовый класс для реализации паттерна "Единица работы".
    /// Отвечает за контроль транзакций и отслеживание изменений в агрегатах через TrackingRepository.
    class AbstractUnitOfWork : public Domain::IUnitOfWork
    {
    public:
        /// repo - Репозиторий, который будет отслеживаться в TrackingRepository.
        explicit AbstractUnitOfWork(Domain::IUpdatableRepository& repo);

        /// Подтверждает изменения.
        void Commit() override
        {
            for (const auto& [product, _] : _tracking.GetSeen())
                if (product->IsModified())
                    _tracking.Add(product);
            _isCommitted = true;
        }

        /// Откатывает изменения.
        void RollBack() override;

        /// Проверяет, были ли изменения зафиксированы.
        bool IsCommitted() const noexcept override;

        /// Возвращает репозиторий для работы с агрегатами-продуктами.
        [[nodiscard]] Domain::IRepository& GetProductRepository() override
        {
            return _tracking;
        }

        /// Возвращает новые сообщения, сгенерированные продуктами
        /// в рамках текущей единицы работы.
        [[nodiscard]] std::vector<Domain::IMessagePtr> GetNewMessages() noexcept override
        {
            std::vector<Domain::IMessagePtr> newMessages;
            for (const auto& [product, _] : _tracking.GetSeen())
            {
                auto messages = product->Messages();
                newMessages.insert(newMessages.end(), messages.begin(), messages.end());
                product->ClearMessages();
            }
            return newMessages;
        }

    private:
        Adapters::Repository::TrackingRepository _tracking;
        bool _isCommitted{false};
    };
}

Компоненты сервисного слоя

Шина сообщений

Шина сообщений (MessageBus) реализует паттерн «издатель–подписчик», принимает сообщение (IMessage) и вызывает соответствующие ему обработчики.

Существуют различия в обработке доменных событий и команд:

1. Команды могут иметь только один обработчик.

2. Исключения в обработчиках команд останавливают дальнейшую обработку.

3. События могут иметь несколько обработчиков.

4. Исключения в обработчиках событий не останавливаю дальнейшую обработку.

• Обработчики событий должны поддерживать концепцию:

    /// Концепция для обработчиков событий конкретного типа.
    template <typename F, typename T>
    concept EventHandlerFor =
        std::derived_from<T, Domain::Events::AbstractEvent> &&
        (
            std::is_invocable_v<F, Domain::IUnitOfWork&, std::shared_ptr<T>> ||
            std::is_invocable_v<F, std::shared_ptr<T>>
        );

• Обработчики команд должны поддерживать концепцию:

    /// Концепция для обработчиков команд конкретного типа.
    template <typename F, typename T>
    concept CommandHandlerFor =
        std::derived_from<T, Domain::Commands::AbstractCommand> &&
        std::is_invocable_v<F, Domain::IUnitOfWork&, std::shared_ptr<T>>;

Точкой входа в обработку сообщений является методы Handle(...).

namespace Allocation::ServiceLayer
{
    /// Шина сообщений для обработки событий и команд.
    class MessageBus
    {
        using EventHandler = std::function<void(Domain::IUnitOfWork&, Domain::Events::EventPtr)>;
        using CommandHandler =
            std::function<void(Domain::IUnitOfWork&, Domain::Commands::CommandPtr)>;

    public:
        static MessageBus& Instance();

        /// Подписывает обработчик на событие конкретного типа.
        /// T - Тип события, производный от Domain::Events::AbstractEvent.
        /// F - Тип обработчика.
        template <typename T, typename F>
        requires EventHandlerFor<F, T>
        void SubscribeToEvent(F&& handler) noexcept
        {
            auto& handlers = _eventHandlers[typeid(T)];
            handlers.emplace_back(
                [h = std::forward<F>(handler)](
                    Domain::IUnitOfWork& uow, Domain::Events::EventPtr event)
                {
                    if constexpr (std::is_invocable_v<F, Domain::IUnitOfWork&, std::shared_ptr<T>>)
                        h(uow, std::static_pointer_cast<T>(event));
                    else if constexpr (std::is_invocable_v<F, std::shared_ptr<T>>)
                        h(std::static_pointer_cast<T>(event));
                });
        }

        /// Устанавливает обработчик для команды конкретного типа.
        /// T - Тип команды, производный от Domain::Commands::AbstractCommand.
        /// F - Тип функции-обработчика.
        template <typename T, typename F>
        requires CommandHandlerFor<F, T>
        void SetCommandHandler(F&& handler) noexcept
        {
            _commandHandlers[typeid(T)] =
                [h = std::forward<F>(handler)](Domain::IUnitOfWork& uow,
                                            Domain::Commands::CommandPtr cmd)
                {
                    h(uow, std::static_pointer_cast<T>(cmd));
                };
        }

        /// Обрабатывает входящее доменное сообщение.
        void Handle(Domain::IMessagePtr message, Domain::IUnitOfWork& uow)
        {
            std::queue<Domain::IMessagePtr> queue;
            queue.push(message);

            while (!queue.empty())
            {
                auto message = queue.front();
                queue.pop();
                if (message->GetType() == Domain::IMessage::Type::Command)
                {
                    if (!_commandHandlers.contains(typeid(*message)))
                        throw std::runtime_error(
                            std::format("The {} command doesn`t have a handler", message->Name()));

                    HandleCommand(uow,
                        std::static_pointer_cast<Domain::Commands::AbstractCommand>(message), queue);
                }
                else if (message->GetType() == Domain::IMessage::Type::Event &&
                     _eventHandlers.contains(typeid(*message)))
                {
                    HandleEvent(
                        uow, std::static_pointer_cast<Domain::Events::AbstractEvent>(message), queue);
                }
            }
        }

        /// Обрабатывает входящее доменное сообщение.
        /// Автоматически создаёт единицу работы SqlUnitOfWork.
        void Handle(Domain::IMessagePtr message);

        /// Очищает все зарегистрированные обработчики событий и команд.
        void ClearHandlers() noexcept;

    private:
        ...

        /// Обрабатывает входящее событие.
        /// uow - Единица работы для обработки события.
        /// event - Доменное событие.
        /// queue - Очередь для новых сообщений.
        void HandleEvent(Domain::IUnitOfWork& uow, Domain::Events::EventPtr event,
            std::queue<Domain::IMessagePtr>& queue) noexcept;
        {
            for (auto& handler : _eventHandlers[typeid(*event)])
            {
                try
                {
                    Loggers::GetLogger()->Debug(std::format("Handling event {} with handler {}",
                        event->Name(), handler.target_type().name()));
                    handler(uow, event);
                    for (auto& newMessage : uow.GetNewMessages())
                        queue.push(newMessage);
                }
                catch (...)
                {
                    Loggers::GetLogger()->Error(
                        std::format("Exception handling event {}", event->Name()));
                }
            }
        }

        /// Обрабатывает входящую команду.
        /// uow - Единица работы для обработки команды.
        /// command - Доменная команда.
        /// queue - Очередь для новых сообщений.
        void HandleCommand(Domain::IUnitOfWork& uow, Domain::Commands::CommandPtr command,
            std::queue<Domain::IMessagePtr>& queue)
        {
            Loggers::GetLogger()->Debug(std::format("handling command {}", command->Name()));
            try
            {
                _commandHandlers.at(typeid(*command))(uow, command);
                for (auto& newMessage : uow.GetNewMessages())
                    queue.push(newMessage);
            }
            catch (...)
            {
                Loggers::GetLogger()->Error(
                    std::format("Exception handling command {}", command->Name()));
                throw;
            }
        }

        std::unordered_map<std::type_index, std::vector<EventHandler>> _eventHandlers;
        std::unordered_map<std::type_index, CommandHandler> _commandHandlers;
    };
}

Пример:

void AddBatch(Domain::IUnitOfWork& uow, std::shared_ptr<Domain::Commands::CreateBatch> message);

void AddAllocationToReadModel(Domain::IUnitOfWork& uow, std::shared_ptr<Domain::Events::Allocated> event);
void PublishAllocatedEvent(Domain::IUnitOfWork& uow);

void Example()
{
    auto& messagebus = ServiceLayer::MessageBus::Instance();
    messagebus.SubscribeToEvent<Domain::Events::Allocated>(AddAllocationToReadModel);
    messagebus.SubscribeToEvent<Domain::Events::Allocated>(PublishAllocatedEvent);
    messagebus.SetCommandHandler<Domain::Commands::CreateBatch>(AddBatch);

    messagebus.Handle(Make<Domain::Events::Allocated>); // вызовет AddAllocationToReadModel и PublishAllocatedEvent
    messagebus.Handle(Make<Domain::Commands::CreateBatch>); // вызовет AddBatch
}

SqlUnitOfWork

SqlUnitOfWork является реализацией паттерна единицы работы для СУБД PosgresSQL.

namespace Allocation::ServiceLayer::UoW
{
    /// Реализация единицы работы для SQL хранилища.
    class SqlUnitOfWork final : public AbstractUnitOfWork
    {
    public:
        /// При создании объекта открывает сессию к БД и начинает транзакцию.
        SqlUnitOfWork()
            : _session(Adapters::Database::DatabaseSessionPool::Instance().GetSession()),
              _repository(_session),
              AbstractUnitOfWork(_repository)
        {
            _session.setTransactionIsolation(Poco::Data::Session::TRANSACTION_REPEATABLE_READ);
            _session.begin();
        }

        /// Откатывает незафиксированные изменения.
        ~SqlUnitOfWork()
        {
            _session.rollback();
        }

        /// Возвращает сессию подключения к базе данных.
        [[nodiscard]] Poco::Data::Session GetSession() noexcept override
        {
            return _session;
        }

        /// Подтверждает внесённые изменения.
        /// После фиксаций изменений запускает новую транзакцию.
        void Commit() override
        {
            AbstractUnitOfWork::Commit();
            _session.commit();
            _session.begin();
        }

        /// Откатывает внесённые изменения.
        /// После отката изменений запускает новую транзакцию.
        void RollBack() override
        {
            _session.rollback();
            AbstractUnitOfWork::RollBack();
            _session.begin();
        }

    private:
        Poco::Data::Session _session;
        Adapters::Repository::SqlRepository _repository;
    };
}

Описание CQRS

В приложении используются две модели хранения данных:

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

2. Модель чтения отвечает за оптимальный доступ к данным.

Модель чтения может быть реализована на любой системе хранения данных.

Внимание: Разделение моделей хранения данных приводит к временной несогласованности между ними. Вопрос глубже рассматривается, например, в книге «Микросервисы. Паттерны, разработка и рефакторинг» Криса Ричардсона.

Связь между моделями реализуется через обработчики событий в шине сообщений:

namespace Allocation::ServiceLayer::Handlers
{
    /// Добавляет в модель чтения распределённую позицию заказа.
    /// uow - Единица работы.
    /// event - Событие "Распределена позиция заказа".
    void AddAllocationToReadModel(
        Domain::IUnitOfWork& uow, std::shared_ptr<Domain::Events::Allocated> event);

    /// Удаляет в модели чтения распределённую позицию заказа.
    /// uow - Единица работы.
    /// event - Событие "Отменено распределение позиции заказа".
    void RemoveAllocationFromReadModel(
        Domain::IUnitOfWork& uow, std::shared_ptr<Domain::Events::Deallocated> event);
}

Модель чтения

В приложении модель чтения представлена отдельной таблицей в базе данных:

CREATE TABLE allocation.allocations_view (
    orderid VARCHAR(255),
    sku VARCHAR(255),
    batchref VARCHAR(255)
);

Эта таблица является проекцией (read model) и синхронизируется моделью команд исключительно через обработчики событий AddAllocationToReadModel и RemoveAllocationFromReadModel.

• при распределении позиции заказа в партии поставок агрегат генерирует событие Domain::Events::Allocated,

• при отмене распределение позиции заказа агрегат генерирует событие Domain::Events::Deallocated,

оба события обрабатываются шиной сообщений и вызывают соответствующие обработчики для обновления модели чтения.

Чтение данных реализовано:

namespace Allocation::ServiceLayer::Views
{
    /// Получает распределённые позиции заказа по идентификатору заказа клиента.
    std::vector<std::pair<std::string, std::string>> Allocations(
        std::string orderid, Domain::IUnitOfWork& uow)
    {
        std::vector<std::pair<std::string, std::string>> results;
        auto session = uow.GetSession();
        Poco::Data::Statement select(session);
        select << "SELECT sku, batchref FROM allocation.allocations_view WHERE orderid = $1",
            Poco::Data::Keywords::use(orderid), Poco::Data::Keywords::now;

        Poco::Data::RecordSet rs(select);
        for (bool more = rs.moveFirst(); more; more = rs.moveNext())
            results.emplace_back(rs["sku"].toString(), rs["batchref"].toString());
        return results;
    }
}

Модель команд

Обработчики доменных команд в шине сообщений являются реализацией взаимодействия с моделью команд.

Пример:

namespace Allocation::ServiceLayer::Handlers
{
    /// Добавляет новую партию заказа.
    /// uow - Единица работы.
    /// command - Команда "Создать партию заказа".
    void AddBatch(Domain::IUnitOfWork& uow, std::shared_ptr<Domain::Commands::CreateBatch> message)
    {
        auto& repo = uow.GetProductRepository();
        auto product = repo.Get(message->sku);
        if (!product)
        {
            product = std::make_shared<Domain::Product>(message->sku);
            repo.Add(product);
        }
        product->AddBatch(
            Allocation::Domain::Batch(message->ref, message->sku, message->qty, message->eta));
        uow.Commit();
    }
}

Слой адаптеров

SQL адаптеры

Адаптеры включают:

• Пул сессий подключений к СУБД.

• Data Mappers для отображения доменных объектов в базу данных и из неё.

• Репозиторий SqlRepository для работы с PostgreSQL СУБД.

Репозиторий

SqlRepository — реализация порта IUpdatableRepository для хранилища PostgreSQL. Он отображает объекты в БД и обратно с помощью Data Mapper-ов.

namespace Allocation::Adapters::Repository
{
    /// Реализация репозитория для работы с PostgreSQL СУБД.
    class SqlRepository final : public Domain::IUpdatableRepository
    {
    public:
        explicit SqlRepository(const Poco::Data::Session& session);

        /// Добавляет новый агрегат-продукт в репозиторий.
        void Add(Domain::ProductPtr product) override;

        /// Возвращает агрегат-продукт по его артикулу.
        [[nodiscard]] Domain::ProductPtr Get(const std::string& sku) override;

        /// Возвращает агрегат-продукт по идентификатору партии включённого в него.
        [[nodiscard]] Domain::ProductPtr GetByBatchRef(const std::string& batchRef) override;

        /// Обновляет агрегат-продукт в репозитории.
        void Update(Domain::ProductPtr product, size_t oldVersion) override;

    private:
        Database::Mapper::ProductMapper _mapper;
    };
}

Используется совместно с TrackingRepository из дочерних классов AbstractUnitOfWork.

В приложении используется принцип мягкого удаление. Созданные ранее агрегаты не удаляются. Удаление партий поставок производится модификацией агрегата.

Data Mapper

Data Mapper-ы отвечают за отображение объектов домена в БД и из неё. Они формируют SQL-запросы для поиска, вставки, удаления и обновления объектов доменов в реляционную БД. Мапперы могут включать друг друга в реализациях.

Для демонстрации продемонстрирован метод ProductMapper::FindBySKU(...):

    /// Маппер для отображения агрегата-продукт в базе данных и обратно.
    class ProductMapper
    {
    public:
        /// session - Сессия подключения к базе данных.
        explicit ProductMapper(const Poco::Data::Session& session);

        /// Находит агрегат-продукт по артикулу.
        [[nodiscard]] Domain::ProductPtr FindBySKU(const std::string& sku) const
        {
            if (sku.empty())
                return nullptr;

            int version;
            Poco::Data::Statement selectProduct(_session);
            selectProduct << R"(
                SELECT version_number
                FROM allocation.products
                WHERE sku = $1
            )",
                useRef(sku), into(version);

            bool found = selectProduct.execute() > 0;
            if (!found)
                return nullptr;

            auto batches = _batchMapper.Find(sku);
            return std::make_shared<Domain::Product>(sku, batches, version, false);
        }

        /// Находит агрегат-продукт по идентификатору партии включённого в него.
        [[nodiscard]] Domain::ProductPtr FindByBatchRef(const std::string& ref) const;

        /// Обновляет агрегат-продукт.
        /// true - успешное обновление, иначе false.
        [[nodiscard]] bool Update(Domain::ProductPtr product, size_t oldVersion);

        /// Сохраняет агрегат-продукт.
        void Insert(Domain::ProductPtr product);

        /// Удаляет агрегат-продукт.
        bool Delete(Domain::ProductPtr product);

    private:
        ...

        mutable Poco::Data::Session _session;
        BatchMapper _batchMapper;
    };

Redis адаптер

Redis адаптер реализует паттерн Publisher и используется как вторичный (ведомый) адаптер для интеграции с внешними системами.

namespace Allocation::Adapters::Redis
{
    /// Публикует события в Redis.
    template <typename T>
        requires std::derived_from<T, Domain::Events::AbstractEvent>
    class RedisEventPublisher
    {
    public:
        RedisEventPublisher() : _connection(RedisConnectionPool::Instance().GetConnection()) {}

        /// Публикует событие в указанный канал.
        void operator()(const std::string& channel, std::shared_ptr<T> event) const
        {
            Poco::JSON::Object json;
            /// формирует JSON на основе аттрибутов события
            for (const auto& [name, value] : GetAttributes<T>(event))
                json.set(name, value);

            std::stringstream ss;
            json.stringify(ss);
            Poco::Redis::Command publish("PUBLISH");
            publish << channel << ss.str();

            try
            {
                /// отправляет событие в канал
                static_cast<Poco::Redis::Client::Ptr>(_connection)->execute<Poco::Int64>(publish);
            }
            catch (const Poco::Exception& e)
            {
                Allocation::Loggers::GetLogger()->Error(
                    std::format("Redis publish failed: {}", e.displayText()));
            }
            catch (...)
            {
                Allocation::Loggers::GetLogger()->Error("Redis publish failed: unknown error");
            }
        }

    private:
        mutable Poco::Redis::PooledConnection _connection;
    };
}

В Domain для каждого события специализирован шаблон функции GetAttributes, который возвращает пары «имя атрибута - значение».

Пример применения:

namespace Allocation::ServiceLayer::Handlers
{
    /// Концепция для отправителей сообщений в канал Redis.
    template <typename T, typename Message>
    concept PublisherSender =
        requires(T t, const std::string& channel, std::shared_ptr<Message> event) {
            { t(channel, event) } -> std::same_as<void>;
        };

    /// Отправитель сообщений в канал Redis.
    template <typename Message, PublisherSender<Message> Publisher>
        requires std::derived_from<Message, Domain::Events::AbstractEvent>
    class PublisherHandler
    {
    public:
        /// publisher - Отправитель сообщений.
        PublisherHandler(Publisher publisher = {}) : _publisher(std::move(publisher)) {}

        /// Публикует сообщение в канал Redis.
        void operator()(std::shared_ptr<Message> event) const
        {
            Allocation::Loggers::GetLogger()->Debug(
                std::format("publishing: channel={}, event={}", "line_allocated", event->Name()));
            _publisher("line_allocated", event);
        }

    private:
        Publisher _publisher;
    };

    /// Публикует событие в Redis "Распределена позиция заказа".
    using PublishAllocatedEvent = PublisherHandler<Domain::Events::Allocated,
    Allocation::Adapters::Redis::RedisEventPublisher<Domain::Events::Allocated>>;

    /// Далее регистрируем PublishAllocatedEvent в шине сообщений как функтор
}

Система уведомлений

Для упрощения реализации система отправки уведомлений по электронной почте заменена заглушкой.

Это пример Stub-реализации адаптера. В реальном приложении он заменяется SMTP-клиентом или API внешнего сервиса.

namespace Allocation::Adapters::Notification
{
    /// Заглушка отправителя email-уведомлений.
    class EmailSenderStub
    {
    public:
        /// Имитирует отправку email-уведомления.
        void operator()(const std::string& to, const std::string& message) const
        {
            Allocation::Loggers::GetLogger()->Debug(
                std::format("Sending email to {}: {}", to, message));
        }
    };
}

Пример применения:

namespace Allocation::ServiceLayer::Handlers
{
    /// Концепция для отправителей уведомлений.
    template <typename T>
    concept NotificationSender = requires(T t, const std::string& to, const std::string& msg) {
        { t(to, msg) } -> std::same_as<void>;
    };

    /// Отправитель уведомлений.
    template <typename Message, NotificationSender Notifier>
        requires std::derived_from<Message, Domain::Events::AbstractEvent>
    class NotificationHandler
    {
    public:
        /// notifier - Отправитель уведомлений.
        NotificationHandler(Notifier notifier = {}) : _notifier(std::move(notifier)) {}

        /// Отправляет уведомление.
        /// В данном примере адрес получателя и текст сообщения захардкожены.
        void operator()(std::shared_ptr<Message> event) const
        {
            _notifier("stock@made.com", std::format("Out of stock for {}", event->sku));
        }

    private:
        Notifier _notifier;
    };

    /// Отправляет уведомление по электронной почте,
    /// по событию "Нет в наличии товара".
    using SendOutOfStockNotification = NotificationHandler<Domain::Events::OutOfStock,
        Allocation::Adapters::Notification::EmailSenderStub>;

    /// Далее регистрируем SendOutOfStockNotification в шине сообщений как функтор
}

Тестирование. Методология TDD

Материал параграфа основан на книгах: «Паттерны разработки на Python: TDD, DDD и событийно-ориентированная архитектура», «Микросервисы. Паттерны разработки и рефакторинга» и «Экстремальное программирование: разработка через тестирование».

Методология TDD

Правила TDD:

1. Новый код пишется только после того, как создан автоматизированный тест, который изначально падает.

2. Любое дублирование кода устраняется как можно скорее.

Эти правила влекут за собой требования к процессу и окружению: быстрый цикл сборки/запуска тестов, ответственность разработчиков за тесты, архитектура с малосвязанными компонентами.

Цикл разработки в TDD:

1. Red — написать тест, который не проходит.

2. Green — реализовать минимально необходимый код, чтобы тест прошёл.

3. Refactor — улучшить структуру и устранить дублирование, сохранив проход тестов.

Цикл повторяется до достижения требуемой функциональности.

Типы тестов

В книге «Микросервисы. Паттерны разработки и рефакторинга» вводятся уровни тестов:

1. Модульные тесты (Unit Tests)

2. Интеграционные тесты (Integration Tests)

3. Компонентные тесты (Component Tests)

4. Сквозные тесты, E2E тесты (End-to-End Tests)

1. Модульные тесты

Модульные тесты применяются для проверки поведения отдельных компонентов приложения. В них проверяются инварианты, граничные условия и другие особенности логики. Важно, что такие тесты не взаимодействуют с внешними системами — например, с СУБД или брокером сообщений.

Гексагональная архитектура позволяет использовать фейковые реализации портов через инъекции зависимостей. Такой подход позволяет изолировать тестируемый код от внешней инфраструктуры.

Пример — фейковый репозиторий:

namespace Allocation::Tests
{
    /// Фейковый репозиторий для тестирования.
    class FakeRepository final : public Domain::IUpdatableRepository
    {
    public:
        FakeRepository() = default;

        /// init - Инициализирующий список продуктов.
        FakeRepository(const std::vector<Domain::ProductPtr>& init)
        {
            for (const auto& prod : init)
                _skuByProduct.insert({prod->GetSKU(), prod});
        }

        /// Добавляет или обновляет продукт в репозитории.
        void Add(Domain::ProductPtr product) override
        {
            _skuByProduct.insert_or_assign(product->GetSKU(), product);
        }

        /// Получает продукт по артикулу.
        [[nodiscard]] Domain::ProductPtr Get(const std::string& SKU) override
        {
            auto it = _skuByProduct.find(SKU);
            return (it != _skuByProduct.end()) ? it->second : nullptr;
        }

        /// Получает продукт по ссылке на партию.
        [[nodiscard]] Domain::ProductPtr GetByBatchRef(const std::string& batchRef) override
        {
            for (const auto& [_, product] : _skuByProduct)
                if (product->GetBatch(batchRef) != std::nullopt)
                    return product;
            return nullptr;
        }

        /// Обновляет продукт (используется с TrackingRepository).
        void Update(Domain::ProductPtr product, int) override
        {
            _skuByProduct.insert_or_assign(product->GetSKU(), product);
        }

        std::unordered_map<std::string, Domain::ProductPtr> _skuByProduct;
    };
}

Фейковая реализация Unit of Work:

namespace Allocation::Tests
{
    /// Фейковая реализация Unit of Work для тестирования.
    class FakeUnitOfWork final : public ServiceLayer::UoW::AbstractUnitOfWork
    {
    public:
        FakeUnitOfWork() : AbstractUnitOfWork(_repo) {}

        /// Получение сессии базы данных.
        /// Возвращает фейковую сессию подключения к базе данных.
        Poco::Data::Session GetSession() noexcept 
        {
            return new FakeSessionImpl("connection_string");
        }

    private:
        FakeRepository _repo;
    };
}

Пример теста:

namespace Allocation::Tests
{
    class Handlers_TestAddBatch : public testing::Test
    {
    public:
        static void SetUpTestSuite()
        {
            ServiceLayer::MessageBus::Instance()
                .SetCommandHandler<Allocation::Domain::Commands::CreateBatch>(
                    ServiceLayer::Handlers::AddBatch);
        }
    };

    TEST_F(Handlers_TestAddBatch, test_for_new_product)
    {
        FakeUnitOfWork uow;
        ServiceLayer::MessageBus::Instance().Handle(
            Make<Domain::Commands::CreateBatch>("b1", "CRUNCHY-ARMCHAIR", 100), uow);

        /// проверка обработчика в изоляции от инфраструктуры
        EXPECT_TRUE(uow.GetProductRepository().Get("CRUNCHY-ARMCHAIR"));
        EXPECT_TRUE(uow.IsCommited());
    }
}

2. Интеграционные тесты

Интеграционные тесты проверяют корректность взаимодействия компонентов системы с внешними элементами — например, с СУБД.
Для подготовки инфраструктуры удобно использовать фикстуры, которые настраивают окружение перед тестами и очищают его по завершению.

Пример фикстуры для работы с СУБД:

/// Фикстура для инициализации БД.
class Database_Fixture : public testing::Test
{
public:
    /// Настраивает пул сессий.
    static void SetUpTestSuite()
    {
        if (auto& sessionPool = Adapters::Database::SessionPool::Instance();
            !sessionPool.IsConfigured())
        {
            auto config = ReadDatabaseConfigurations();
            sessionPool.Configure(config);
        }
    }

protected:
    /// Выполняется перед выполнением каждого теста
    void SetUp() override
    {
        _session = Adapters::Database::SessionPool::Instance().GetSession();
        _session.begin();
    }

    /// Выполняется по завершению каждого теста
    void TearDown() override
    {
        try
        {
            _session.rollback();
        }
        catch (...)
        {
        }
    }

    Poco::Data::Session _session{Adapters::Database::SessionPool::Instance().GetSession()};
};

Пример интеграционного теста:

namespace Allocation::Tests
{
    TEST_F(Database_Fixture, test_get_by_batchref)
    {
        Adapters::Repository::SqlRepository repo(_session);

        Domain::Batch b1("b1", "sku1", 100);
        Domain::Batch b2("b2", "sku1", 100);
        Domain::Batch b3("b3", "sku2", 100);

        auto p1 = std::make_shared<Domain::Product>("sku1", std::vector<Domain::Batch>{b1, b2});
        auto p2 = std::make_shared<Domain::Product>("sku2", std::vector<Domain::Batch>{b3});

        /// Сохраняем агрегаты в БД
        repo.Add(p1);
        repo.Add(p2);

        /// Проверяем работу SqlRepository
        /// Загружаем ранее сохранённые агрегаты 
        EXPECT_EQ(repo.GetByBatchRef("b2"), p1);
        EXPECT_EQ(repo.GetByBatchRef("b3"), p2);
    }
}

Так как в фикстуре сессия базы данных открывается, но изменения не фиксируются (rollback в TearDown), изменения откатываются в БД.

3. Компонентные тесты

Компонентные тесты применяются для проверки самодостаточных единиц приложения.
В контексте микросервисной архитектуры такой единицей является отдельный сервис.

При реализации компонентных тестов взаимодействующие с тестируемым сервисом системы заменяются заглушками или упрощёнными эмуляторами.

Описываемом проекте разрабатывается один сервис, поэтому компонентные тесты как отдельный уровень тестирования не применяются.

4. Сквозные тесты

Сквозные тесты (end-to-end, E2E) применяются для проверки работы всего приложения.
На этом уровне приложение рассматривается как «чёрный ящик»: тесты взаимодействуют с ним через API, проверяют корректность поведения с точки зрения внешнего клиента.

В проекте для написания e2e-тестов используется ЯП Python.

Пример Е2Е теста:

def test_happy_path_returns_202_and_batch_is_allocated():
    orderid = random_orderid()
    sku, othersku = random_sku(), random_sku("other")
    earlybatch = random_batchref(1)
    laterbatch = random_batchref(2)
    otherbatch = random_batchref(3)

    api_client.post_to_add_batch(laterbatch, sku, 100, "2011-01-02T00:00:00")
    api_client.post_to_add_batch(earlybatch, sku, 100, "2011-01-01T00:00:00")
    api_client.post_to_add_batch(otherbatch, othersku, 100, None)

    r = api_client.post_to_allocate(orderid, sku, qty=3)
    assert r.status_code == 202

    r = api_client.get_allocation(orderid)
    assert r.ok
    assert r.json() == [{"sku": sku, "batchref": earlybatch}]

Количество тестов

Для описания распределения количества тестов по уровням используют геометрические фигуры. Наиболее распространены:

• Пирамида

• Трапеция

• Перевёрнутая пирамида

Уровни тестов располагаются послойно: начиная с модульных тестов в основании и заканчивая E2E-тестами на вершине.

Каждая фигура отражает пропорцию тестов в системе:

• Пирамида — основное количество приходится на модульные тесты, меньше — на интеграционные, ещё меньше — на компонентные и совсем немного на E2E.

• Трапеция — упор делается на интеграционные тесты, при этом модульных относительно меньше.

• Перевёрнутая пирамида — большая часть тестов пишется на уровне E2E, при этом модульные тесты практически отсутствуют.

Перевёрнутая пирамида считается антипаттерном, так как такие тесты:

• выполняются медленно,

• широко охватывают систему,

• сложно поддерживаются.

В проекте придерживаемся пирамидальной стратегии: основной упор делаем на модульные тесты, дополняя их интеграционными и минимальным количеством E2E

Заключение

В статье была рассмотрена реализация сервиса на C++ с опорой на идеи книги «Паттерны разработки на Python: TDD, DDD и событийно-ориентированная архитектура». Обсуждены ключевые слои приложения, архитектурные решения, подходы к тестированию и особенности практической реализации.

В дальнейшем планируется перенести проект на фреймворк userver ^[15].

Список литературы

1. Гарри Персиваль, Боб Грегори. Паттерны разработки на Python: TDD, DDD и событийно-ориентированная архитектура.

2. Мартин Фаулер. Шаблоны корпоративных приложений.

3. Крис Ричардсон. Микросервисы. Паттерны, разработка и рефакторинг.

4. Клаус Игльбергер. Проектирование программ на C++. Принципы и паттерны.

5. Влад Хононов. Изучаем DDD — предметно-ориентированное проектирование.

6. Кент Бек Экстремальное программирование: разработка через тестирование

Благодарность

Автор благодарит организацию «Тис-Центр», в которой работает, за поддержку инициативы, а также коллег, чьи советы помогли в подготовке материала.

Отдельная благодарность:

• Номхоеву Владимиру Николаевичу

• Галчину Дмитрию Андреевичу