Сегодня мы выкатили очередную версию so_5_extra -- 1.0.1, а накануне обновили SObjectizer до версии 5.5.19.3. Доступно все, как обычно, на SourceForge: здесь so_5_extra, здесь SObjectizer.
В so_5_extra-1.0.1 добавлена всего одна фича: это collecting_mbox.
Смысл collecting_mbox в том, что иногда агенту X нужно получить строго N сообщений типа M от других агентов. Ну вот, например, стартует родительский агент-менеджер и создает пять нужных ему дочерних коопераций. Каждая дочерняя кооперация тратит какое-то время на свои начальные действия (например, загружается конфигурация, устанавливается подключение к БД, инициализируется внешнее устройство и т.д.), после чего отсылает родителю сигнал о том, что дочерняя кооперация готова. Соответственно, родительскому агенту нужно получить все пять сигналов о готовности дочерних коопераций, лишь после этого родительский агент сможет продолжать работать.
Без использования collecting_mbox эта задача решается за счет заведения счетчика внутри родительского агента. Т.е. чего-то вроде:
|
// Сигнал о том, что очередная дочерняя кооперация успешно стартовала. struct child_started_t final : public so_5::signal_t {}; class parent_agent_t final : public so_5::agent_t { // Счетчик дочерних коопераций, который подтвердили свой // успешный старт. unsigned int m_child_started = 0; // Реакция на поступление очередного сигнала о том, что дочерняя // кооперация стартовала. void on_child_started(mhood_t<child_started_t>) { ++m_child_started; if( 5 == m_child_started ) ... // Можно начинать основную работу. } ... }; |
А вот при использовании collecting_mbox вручную считать отдельные сигналы не нужно. Смысл collecting_mbox в том, что мы создаем новый mbox для определенного типа сообщений/сигналов и говорим, сколько сообщений/сигналов туда должно прийти. Отдаем этот новый mbox своим дочерним агентам. Когда дочерний агент завершает свои начальные действия, он отсылает сигнал в этот новый mbox. Новый mbox подсчитывает, сколько всего сигналов он получил. Когда получает все, то mbox сам отсылает специальное сообщение messages_collected_t на mbox родительского агента.
Простой пример
Вот как описанная выше ситуация может выглядеть в коде (это реальный пример mboxes/collecting_mbox/simple из состава so_5_extra):
|
#include <so_5_extra/mboxes/collecting_mbox.hpp> #include <so_5/all.hpp> // Сигнал, который отсылают дочерние агенты при своей готовности. struct child_started_t final : public so_5::signal_t {}; // Простейший дочений агент, единственной задачей которого является // отсылка сигнала child_started_t после начала работы. class child_t final : public so_5::agent_t { public: child_t(context_t ctx, so_5::mbox_t ready_mbox) : so_5::agent_t(std::move(ctx)) , m_ready_mbox(std::move(ready_mbox)) {} virtual void so_evt_start() override { // Стартовали, можно отослать сигнал родителю. so_5::send<child_started_t>(m_ready_mbox); } private: // На этот mbox нужно будет отсылат сигнал о готовности. const so_5::mbox_t m_ready_mbox; }; // Тип агента-родителя. class sample_performer_t final : public so_5::agent_t { // Тип, который нужен для накопления сигналов child_started_t. using child_started_mbox_t = so_5::extra::mboxes::collecting_mbox::mbox_template_t<child_started_t>; public: // В конструктор передается количество дочерних агентов, // которые нужно создать. // Важно, что это значение для агента становится известно только // в run-time. Это влияет на определение типа child_started_mbox_t. sample_performer_t(context_t ctx, std::size_t child_count) : so_5::agent_t(std::move(ctx)) , m_child_count(child_count) { so_subscribe_self() // Сразу же подписываемся на уведомление о том, что // все дочерние агенты прислали child_started_t. .event(&sample_performer_t::on_all_child_started); } virtual void so_evt_start() override { // Создаем экземпляр collecting_mbox-а. // Для этого используется статический метод make из // child_started_mbox_t. auto ready_mbox = child_started_mbox_t::make( // SObjectizer Environment, в котором ведется работа. // Необходим для того, чтобы новый mbox получил уникальный ID. so_environment(), // На этот mbox затем collecting mbox пришет свое уведомление // о том, что нужное количество сигналов получено. so_direct_mbox(), // При создании задаем количество ожидаемых сигналов. // Затем это количество мы уже не сможем поменять. m_child_count); // Всех дочерних агентов запускаем на отдельном thread_pool-диспетчере. auto tp_disp = so_5::disp::thread_pool::create_private_disp( so_environment(), 3 ); for(std::size_t i = 0; i != m_child_count; ++i) { introduce_child_coop(*this, tp_disp->binder(so_5::disp::thread_pool::bind_params_t{}), [&ready_mbox](so_5::coop_t & coop) { // При создании дочернего агента указываем новый mbox // в качестве получателя сигнала child_started_t. coop.make_agent<child_t>(ready_mbox); }); } std::cout << "All children agents are created" << std::endl; } private: // Количество дочерних агентов, которые предстоит создать. const std::size_t m_child_count; // Реакция на получение уведомления о том, что все дочерние агенты // стартовали. void on_all_child_started( // Это уведомление поступает к нам в виде сообщения messages_collected_t // из типа child_started_mbox_t. mhood_t<child_started_mbox_t::messages_collected_t>) { std::cout << "All children agents are started" << std::endl; so_deregister_agent_coop_normally(); } }; |
Подробнее о collecting_msg
Принцип работы
Экземпляр collecting_mbox получает сообщения определенного типа и накапливает их в своем внутреннем буфере. Когда collecting_mbox накопит N сообщений (где N задается программистом), он отсылает специальное сообщение messages_collected_t в указанный программистом целевой почтовый ящик. В этом новом экземпляре messages_collected_t содержится N накопленных ранее сообщений. После чего collecting_mbox накапливает следующие N сообщений, вновь отсылает новый экземпляр messages_collected_t. Затем снова накапливает N сообщений и т.д.
Возможно накопление только одного типа сообщений. Попытка отослать в collecting_mbox сообщение любого другого типа будет приводить к выбросу исключения.
Collecting_mbox можно использовать только для отсылки сообщений. Нельзя выполнять синхронные запросы с использованием collecting_mbox (т.е. нельзя передавать collecting_mbox в вызовы request_value и request_future -- это будет приводить к выбросу исключений).
Создание collecting_mbox
Для того, чтобы получить collecting_mbox, нужно сделать две вещи.
Во-первых, нужно настроить под свои нужды шаблон mbox_template_t из пространства имен so_5::extra::mboxes::collecting_mbox. Этот шаблон получает три параметра, лишь один из которых обязателен:
- COLLECTING_MSG определяет тип сообщений/сигналов, которые нужно накапливать. В примере выше это был тип сигнала child_started_t;
- TRAITS определяет тип дополнительных свойств mbox-а, зависящих от количества накапливаемых сообщений. По умолчанию в качестве TRAITS используется тип runtime_size_traits_t, который предназначен для случая, когда количество накапливаемых сообщений известно только в run-time. На эту тему далее чуть подробнее;
- LOCK_TYPE определяет какой тип будет у мьютекса для защиты данных в mbox-е от многопоточного доступа. По умолчанию это std::mutex. Для однопоточных вариантов SObjectizer-а можно использовать so_5::null_mutex_t. Можно подсунуть и какой-нибудь собственный тип, который можно использовать совместно с std::lock_guard.
Обычно точный тип collecting_mbox-а задается посредством нужного typedef-а. Что-то вроде:
|
using my_msg_mbox_type = so_5::extra::mboxes::collecting_mbox::mbox_template_t< // Задаем только тип накапливаемых сообщений. // Остальные параметры остаются неизменными. my_msg>; using another_mbox_type = so_5::extra::mboxes::collecting_mbox::mbox_template_t< // Тип сообщения. another_msg_type, // Говорим о том, что количество накапливаемых сообщений будет // известно только в run-time. so_5::extra::mboxes::collecting_mbox::runtime_size_traits_t, // Задаем null_mutex вместо std::mutex. // Т.е. mbox-ы этого типа будут пригодны к использованию только // в рамках однопоточного окружения. so_5::null_mutex_t>; |
Подобные typedef-ы очень сильно упрощают жизнь при создании mbox-ов и при подписке на сообщение messages_collected_t.
Во-вторых, экземпляр collecting_mbox-а нужно создать. Делается это посредством статического метода make шаблонного класса mbox_template_t. Список параметров этого метода зависит от значения параметра TRAITS для mbox_template_t. Если в качестве TRAITS используется runtime_size_traits_t, то у mbox_template_t::make будет следующий формат:
|
so_5::mbox_t make( // В каком SObjectizer Environment идет работа. so_5::environment_t & env, // Куда прислать уведомление о накопленных сообщениях. const so_5::mbox_t & target_mbox, // Сколько именно сообщений нужно накапливать. std::size_t messages_to_collect); |
Если в качестве TRAITS используется constexpr_size_traits_t, то формат mbox_template_t::make меняется, параметр messages_to_collect становится уже не нужным:
|
so_5::mbox_t make( // В каком SObjectizer Environment идет работа. so_5::environment_t & env, // Куда прислать уведомление о накопленных сообщениях. const so_5::mbox_t & target_mbox); |
Если в качестве TRAITS используется еще какой-то тип, то формат mbox_template_t::make будет определяться этим типом. Пока в so_5_extra входит всего два готовых типа, которые могут использоваться в качестве параметра TRAITS: runtime_size_traits_t и constexpr_size_traits_t.
Внутри шаблона mbox_template_t определяется имя messages_collected_t. Это имя определяет тип сообщения, которое будет отсылаться в target_mbox когда collecting_mbox накопит заданное количество сообщений/шаблонов. Реальный тип этого сообщения зависит от нескольких факторов:
- если COLLECTING_MSG задает тип сигнала, то messages_collected_t будет определять сообщение следующего вида:
Т.е., если collecting_mbox накапливает сигналы, то в messages_collected_t не будет никакой другой информации, кроме общего количества накопленных сигналов;class messages_collected_t final : public so_5::message_t
{
...
public:
std::size_t size();
}; - если COLLECTING_MSG задает тип сообщения, то messages_collected_t будет определять сообщение следующего вида:
Т.е. в messages_collected_t будут находиться все накопленные сообщения и к ним можно будет обратиться посредством методов with_nth, for_each и for_each_with_index. Но конкретный тип messages_collected_t и в этом случае будет зависеть от типа TRAITS.class messages_collected_t final : public so_5::message_t
{
some_container_t m_collected_messages;
...
public:
std::size_t size() const;
template< typename F >
decltype(auto) with_nth( std::size_t index, F && f ) const;
template< typename F >
void for_each( F && f ) const;
template< typename F >
void for_each_with_index( F && f ) const;
};
Подробнее про параметр TRAITS
Самый хитрый параметр шаблона mbox_template_t -- это TRAITS. В реализации collecting_mbox-а были специально разделены случаи, когда количество накапливаемых сообщений известно в compile-time, и когда оно известно только в run-time.
Если количество накапливаемых сообщений известно в compile-time, то в реализации соответствующего messages_collected_t для хранения накопленных сообщений используется std::array, размер которого задается в compile-time.
Если же количество накапливаемых сообщения в compile-time не известно и становится доступно только в run-time, то в соответствующем messages_collected_t для хранения накопленных сообщений используется std::vector. Соответственно, с необходимостью динамически выделять память под его содержимое во время исполнения.
По умолчанию в качестве значения TRAITS используется runtime_size_traits_t. Это означает, что при создании mbox-а нужно будет указывать количество накапливаемых сообщений, а для их хранения в сообщении messages_collected_t будет использоваться std::vector.
Если же количество накапливаемых значений известно в compile-time, то можно использовать в качестве значения TRAITS тип constexpr_size_traits_t. Например:
|
using my_mbox_type = so_5::extra::mboxes::collecting_mbox::mbox_template_t< // Тип сообщения. my_msg_type, // Говорим о том, что количество накапливаемых сообщений нам // известно уже в compile-time. // Собираем всего по 5 сообщений. so_5::extra::mboxes::collecting_mbox::constexpr_size_traits_t<5>>; |
Доступ к накопленным сообщениям
Если collecting_mbox используется для накопления сообщений, а не сигналов, то все накопленные экземпляры сообщений доступны для программиста внутри messages_collected_t. Получить доступ к накопленным сообщениям можно посредством методов with_nth, for_each и for_each_with_index.
Метод with_nth дает доступ к i-ому сообщению по индексу. На вход with_nth принимает индекс (от 0 до size()-1) и лямда-функцию/функтор. Эта лямбда-функция вызывается внутри with_nth и ей в качестве единственного параметра отдается mhood_t<COLLECTING_MSG>. Возвращенное лямбдой значение возвращается как результат работы with_nth. Например:
|
void my_agent::on_collected_msg(mhood_t<my_mbox_type::messages_collected_t> cmd) { auto v = cmd->with_nth([](mhood_t<my_msg> m) { return m->some_value(); }); ... } |
Метод for_each предназначен для обхода всех накопленных сообщений. На вход for_each принимает лямбда-функцию/функтор такого же формата, как и with_nth: единственный аргумент типа mhood_t<COLLECTING_MSG>. Эта лямбда-функция будет вызвана для каждого сообщения внутри messages_collected_t. Все, что возвращается лямбда-функцией, игнорируется, т.к. сама for_each ничего не возвращает. Пример использования for_each:
|
void my_agent::on_collected_msg(mhood_t<my_mbox_type::messages_collected_t> cmd) { std::vector<std::string> names; names.reserve(cmd->size()); cmd->for_each([&names](mhood_t<my_msg> m) { names.push_back(m->some_value()); }); ... } |
Метод for_each_with_index так же предназначен для обхода всех накопленных сообщений, но при этом вместе с очередным сообщением в лямбда-функцию/функтор передается так же и порядковый номер сообщения. Т.о. у лямбда-функции два аргумента: первый -- это индекс типа std::size_t, второй -- это mhood_t<COLLECTING_MSG>. Возвращаемое значение лямбда-функции игнорируется, for_each_with_index ничего не возвращает. Пример использования:
|
void my_agent::on_collected_msg(mhood_t<my_mbox_type::messages_collected_t> cmd) { std::vector<std::string> names(cmd->size()); cmd->for_each_with_index( [&names](std::size_t index, mhood_t<my_msg> m) { names[index] = m->some_value(); }); ... } |
collecting_mbox и mutable_msg
Для того, чтобы с помощью collecting_mbox собирать мутабельные сообщения, необходимо выполнение трех условий:
Во-первых, в качестве параметра COLLECTING_MSG следует указывать тип, заключенный внутрь маркера so_5::mutable_msg:
|
using my_mutable_mbox_type = so_5::extra::mboxes::collecting_mbox::mbox_template_t< // Тип сообщения. Явно указывается, что это мутабельное сообщение. so_5::mutable_msg<my_msg_type>>; |
Во-вторых, в качестве target_mbox-а следует указывать только multi-producer/single-consumer mbox. Например, direct-mbox конкретного агента-получателя. Или же mchain. Попытка задать multi-producer/multi-consumer mbox в качестве целевого, будет приводить к выбросу исключения при создании mbox-а.
В-третьих, messages_collected_t будет отсылаться в target_mbox как so_5::mutable_msg<messages_collected_t>. Именно в таком виде и следует подписываться на messages_collected_t:
|
void my_agent::so_define_agent() override { so_subscribe_self().event(&my_agent::on_collected_msg); ... } void my_agent::on_collected_msg( mhood_t<so_5::mutable_msg<my_mutable_mbox_type::messages_collected_t>> cmd) { ... } // Или же: void my_agent::on_collected_msg( mutable_mhood_t<my_mutable_mbox_type::messages_collected_t> cmd) { ... } |
Так же нужно учитывать, что в лямбда-функции, которые передаются в методы with_nth, for_each, for_each_with_index, накопленные сообщения будут приходить в виде mhood_t<mutable_msg<MSG>>. Т.е.:
|
void my_agent::on_collected_msg( mutable_mhood_t<my_mutable_mbox_type::messages_collected_t> cmd) { cmd->with_nth([](mutable_mhood_t<my_msg_type> m) { ... }); } |
Более сложный пример
Рассмотрим более сложный пример использования collecting_mbox. Предположим, что нам нужно хранить информацию о книгах, но в разделенном виде: каждая часть информации хранится в своей шарде. Т.е. один агент-шарда хранит только автора книги, второй агент хранит заголовок книги, третий агент хранит краткое описание. Для того, чтобы сохранить описание книги нам нужно отослать сообщение с описанием книги всем агентам-шардам и дождаться от каждого из них подтверждение о том, что книга сохранена. Аналогично, для того, чтобы собрать полное описание книги, нужно сделать запросы к каждому из агентов-шард запрос, после чего дождаться, пока каждый из агентов-шард ответит.
Вот как это может выглядеть в коде.
Сперва определим все сообщения и сопутствующие инструменты. Начнем с сопутствующих инструментов
|
// Эти значения будут указывать, какая часть описания книги нужна. enum class field_id_t { author, title, summary }; // Описание одной книги. struct book_description_t { std::string m_author; std::string m_title; std::string m_summary; }; // Набор специализаций функции get, которая из описания книги // извлекает нужную часть. template<field_id_t Field> std::string get(const book_description_t &); template<> std::string get<field_id_t::author>(const book_description_t & b) { return b.m_author; } template<> std::string get<field_id_t::title>(const book_description_t & b) { return b.m_title; } template<> std::string get<field_id_t::summary>(const book_description_t & b) { return b.m_summary; } |
А теперь сами сообщения, которые будут использоваться в обмене между агентами-шардами и остальным миром:
|
// Команда для сохранения описания очередной книги. struct store_book_t : public so_5::message_t { // Уникальный ID для книги. По этому ID книга затем может быть запрошена. const int m_key; // Описание новой книги. const book_description_t m_book; // Почтовый ящик для подтверждения о том, что книга создана. const so_5::mbox_t m_ack_to; store_book_t(int key, book_description_t book, so_5::mbox_t ack_to) : m_key(key), m_book(std::move(book)), m_ack_to(std::move(ack_to)) {} }; // Ответное сообщение с подтверждением того, что книга создана. struct store_book_ack_t : public so_5::message_t { // Уникальный ID созданной книги. const int m_key; store_book_ack_t(int key) : m_key(key) {} }; // Запрос описания книги. struct request_data_t : public so_5::message_t { // Уникальный ID запрашиваемой книги. const int m_key; // Почтовый ящик, на который данные о книге нужно прислать. const so_5::mbox_t m_reply_to; request_data_t(int key, so_5::mbox_t reply_to) : m_key(key), m_reply_to(std::move(reply_to)) {} }; // Сообщение от агента-шарды с той информацией о книге, которая у шарды есть. struct data_t : public so_5::message_t { // Уникальный ID запрошенной книги. const int m_key; // Какая именно часть описания книги лежит в m_field. const field_id_t m_field; // Та часть описания, которая была у шарды. const std::string m_data; data_t(int key, field_id_t field, std::string data) : m_key(key), m_field(field), m_data(std::move(data)) {} }; |
Теперь можно определить тип агента-шарды. Это будет шаблон, который параметризуется конкретным field_id:
|
template< // Параметр шаблона задает что именно будет храниться агентом: // автор, заголовок или краткое описание. field_id_t Field> class shard_t final : public so_5::agent_t { public: shard_t(context_t ctx, so_5::mbox_t command_mbox) : so_5::agent_t(std::move(ctx)) { // Команды агенту будут поступать из специального почтового ящика. // Всего агенту потребуется реагировать на две команды, поэтому // подписываются два обработчика событий. so_subscribe(command_mbox) .event(&shard_t::on_store_book_t) .event(&shard_t::on_request_data); } private: // Вся информация хранится в ассоциативном контейнере с доступом // по уникальному ID книги. std::map<int, std::string> m_data; void on_store_book_t(mhood_t<store_book_t> cmd) { // При получении команды на сохранение новой книги сохраняем у себя // часть описания книги и отсылаем подтверждение. m_data[cmd->m_key] = get<Field>(cmd->m_book); so_5::send<store_book_ack_t>(cmd->m_ack_to, cmd->m_key); } void on_request_data(mhood_t<request_data_t> cmd) { // При запросе отсылаем ту часть, которая хранится у агента. so_5::send<data_t>(cmd->m_reply_to, cmd->m_key, Field, m_data[cmd->m_key]); } }; |
Ну и теперь можно показать агента, который общается с агентами-шардами используя функциональность collecting_mbox-ов.
Агент sample_performer_t при своем старте отсылает несколько команд store_book. Когда на очередную команду store_book он получает подтверждение (в виде сообщения store_book_ack) он запрашивает описание сохраненной книги. После того, как все описания будут получены, агент sample_performer_t завершает работу примера.
Прежде чем показать код sample_performer-а, нужно заострить внимание на паре моментов:
- поскольку количество агентов-шард у нас известно заранее и в процессе работы примера оно измениться не может в принципе, то используются collecting_mbox-ы со свойствами constexpr_size_traits_t;
- ответы на команды store_book и request_book запросто могут перемешаться. Например, агент-шарда может получить два сообщения store_book подряд и сразу же ответить на них store_book_ack. И только после этого эти две первые store_book дойдут до двух остальных агентов-шард. Соответственно, нам нужно не просто накапливать ответные store_book_ack и data, но и группировать их по уникальному ID книги. Если мы этого не сделаем, то запросто можем получить messages_collected_t с тремя экземплярами сообщения data_t внутри, однако каждый экземпляр будет относиться к разным книгам! Для того, чтобы избавиться от такой проблемы агент sample_performer создает новый collecting_mbox для каждой своей команды. Получается, что ответы на конкретную команду будут идти в отдельный mbox и, поэтому, эти ответы не будут перемешиваться. Создание нового collecting_mbox-а -- это дешевая операция, да и уничтожается collecting_mbox автоматически после того, как перестает использоваться.
Ну а теперь сам код агента sample_performer:
|
class sample_performer_t final : public so_5::agent_t { // Количество книг, которые используются в примере. static constexpr std::size_t total_books = 3; public: // Управляющий mbox, через который команды отсылаются агентам-шардам // создается снаружи и передается через конструктор. sample_performer_t(context_t ctx, so_5::mbox_t command_mbox) : so_5::agent_t(std::move(ctx)) , m_command_mbox(std::move(command_mbox)) { // Нам нужно ловить всего два типа сообщений, которые будут // прилетать на персональный почтовый ящик агента. so_subscribe_self() .event(&sample_performer_t::on_store_ack) .event(&sample_performer_t::on_data); } virtual void so_evt_start() override { // В начале своей работы регистрируем описания нескольких книг. std::array<book_description_t, total_books> books{{ { "Miguel De Cervantes", "Don Quixote", "The story of the gentle knight and his servant Sancho Panza has " "entranced readers for centuries. " }, { "Jonathan Swift", "Gulliver's Travels", "A wonderful satire that still works for all ages, despite the " "savagery of Swift's vision." }, { "Stendhal", "The Charterhouse of Parma", "Penetrating and compelling chronicle of life in an Italian " "court in post-Napoleonic France." } }}; int key = 0; for( const auto & b : books ) { so_5::send<store_book_t>(m_command_mbox, key, b, // Под каждую команду создаем новый collecting_mbox для // накопления ответов. Накопленные ответы будут отсылаться // на персональный mbox агента. store_ack_mbox_t::make(so_environment(), so_direct_mbox())); ++key; } } private: // Определяем тип collecting_mbox-а для подтверждений сохранения новой книги. using store_ack_mbox_t = so_5::extra::mboxes::collecting_mbox::mbox_template_t< store_book_ack_t, // Количество накапливаемых ответов известно в compile-time, // поэтому используем соответствующий trait. so_5::extra::mboxes::collecting_mbox::constexpr_size_traits_t<3>>; // Определяем тип collecting_mbox-а для ответов на запрос информации о книге. using data_mbox_t = so_5::extra::mboxes::collecting_mbox::mbox_template_t< data_t, so_5::extra::mboxes::collecting_mbox::constexpr_size_traits_t<3>>; // Почтовый ящик, через который отдаются команды агентам-шардам. const so_5::mbox_t m_command_mbox; std::size_t m_books_received = 0; void on_store_ack(mhood_t<store_ack_mbox_t::messages_collected_t> cmd) { // У всех накопленных сообщений внутри одинаковый ID. // Берем его значение из первого сообщения. const auto key = cmd->with_nth(0, [](auto m) { return m->m_key; }); std::cout << "Book with key=" << key << " is stored" << std::endl; // Раз книга успешно сохранена, то можно запросить ее описание. so_5::send<request_data_t>(m_command_mbox, key, // Для каждого запроса создаем новый collecting_mbox. data_mbox_t::make(so_environment(), so_direct_mbox())); } void on_data(mhood_t<data_mbox_t::messages_collected_t> cmd) { // Опять же, у всех сообщений внутри одинаковый ID. // Чтобы получить его достаточно обратиться к первому сообщению. const auto key = cmd->with_nth(0, [](auto m) { return m->m_key; }); // Каждый агент-шарда прислал свою часть описания книги. // Для того, чтобы собрать полное описание нам потребуется пройтись // по всем накопленным сообщениям. Для чего воспользуемся for_each-ем. book_description_t book; cmd->for_each([&book](auto m) { // Тут придется понять, что за часть к нам пришла. if(field_id_t::author == m->m_field) book.m_author = m->m_data; else if(field_id_t::title == m->m_field) book.m_title = m->m_data; else if(field_id_t::summary == m->m_field) book.m_summary = m->m_data; }); std::cout << "Book with key=" << key << " is {" << book.m_author << ", '" << book.m_title << "', " << book.m_summary << "}" << std::endl; // Если получили все книги, то пример можно завершить. ++m_books_received; if(total_books == m_books_received) so_deregister_agent_coop_normally(); } }; |
Заключение
У нас у самих пока еще нет большого опыта работы с collecting_mbox, поэтому первая реализация может казаться примитивной и ограниченной. Отчасти так и есть, поскольку хотелось сделать лишь самый базовый функционал, который казался очевидным. А дальше уже посмотреть на использование collecting_mbox-а и расширять его на основании набитых в реальных проектах шишек.
Так что если кто-то видит какие-то проблемы в первой версии collecting_mbox или у кого-то есть идеи, как это можно расширить и углубить, то мы будем рады услышать ваше мнение дабы учесть его в дальнейшем.
Комментариев нет:
Отправить комментарий