Sophiar2/tests/utility/coro_signal_group.cpp

#define BOOST_TEST_DYN_LINK

#include "core/sophiar_obj.hpp"
#include "utility/coro_signal_group.hpp"
#include "utility/debug_utility.hpp"

#include <boost/asio/co_spawn.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/this_coro.hpp>
#include <boost/asio/use_awaitable.hpp>
#include <boost/test/unit_test.hpp>

#include <chrono>
#include <iostream>
#include <vector>

using boost::asio::awaitable;
using boost::asio::co_spawn;
using boost::asio::detached;
using boost::asio::use_awaitable;

using namespace sophiar;
using namespace std::chrono_literals;

awaitable<void> test_coro_signal_group_1() {
    coro_signal2 sig_a, sig_b;

    coro_signal_any_group group_any;
    coro_signal_all_group group_all;

    group_any.add_watcher(sig_a.new_watcher());
    group_any.add_watcher(sig_b.new_watcher());
    group_all.add_watcher(sig_a.new_watcher());
    group_all.add_watcher(sig_b.new_watcher());

    int cnt_any = 0, cnt_all = 0;

    auto worker_any = make_infinite_coro_worker(
            [&,
                    watcher = group_any.new_watcher()]() mutable
                    -> awaitable<bool> {
                co_await watcher.coro_wait(false);
                ++cnt_any;
                co_return true;
            });
    auto worker_all = make_infinite_coro_worker(
            [&,
                    watcher = group_all.new_watcher()]() mutable
                    -> awaitable<bool> {
                co_await watcher.coro_wait(false);
                ++cnt_all;
                co_return true;
            });

    worker_any->run();
    worker_all->run();

    group_any.start();
    group_all.start();

    sig_a.try_notify_all();
    co_await coro_sleep(10ms);
    BOOST_TEST(cnt_any == 1);
    BOOST_TEST(cnt_all == 0);

    sig_b.try_notify_all();
    co_await coro_sleep(10ms);
    BOOST_TEST(cnt_any == 2);
    BOOST_TEST(cnt_all == 1);

    sig_a.try_notify_all();
    sig_b.try_notify_all();
    co_await coro_sleep(10ms);
    BOOST_TEST(cnt_any == 3);
    BOOST_TEST(cnt_all == 2);

    worker_any->cancel();
    worker_all->cancel();
    co_await worker_any->coro_wait_stop();
    co_await worker_all->coro_wait_stop();
    co_await group_any.stop();
    co_await group_all.stop();

    co_return;
}

BOOST_AUTO_TEST_CASE(test_coro_signal_group) {
    initialize({});

    co_spawn(*global_context, test_coro_signal_group_1(), detached);
    global_context->run();
}