gistfile1.txt

net::awaitable< void >
monitor_subscription(async::st::stop_token                      stop,
                     util::printable_base const &               ident,
                     entity::entity_service                     owner,
                     entity::universal_key const &              key,
                     async::st::ready_latch_impl &              ready_latch,
                     trade::snapshot_condition &                condition,
                     std::function< void(json::value const &) > update)
try
{
    spdlog::trace("{}::{} entry", ident, __func__);

    // premable
    auto conn = trade::deribit_connection(owner);

    auto to_channel = [&key] {
        auto s = to_deribit(key);
        return json::string(s.begin(), s.end());
    };

    auto subs     = co_await conn.public_subscribe(to_channel());
    auto stopconn = stop.connect([&] { subs.disconnect(); });

    while (!stop.stopped())
    {
        auto upsnap = co_await subs.consume();
        spdlog::debug(
            "{}::{} received data {}", ident, __func__, print(upsnap));
        try
        {
            condition.reset_good();
            update(upsnap);
        }
        catch (...)
        {
            spdlog::error("{}::{} error {}", ident, __func__, explain());
            condition.reset_error(trade::error_info(to_string(to_string(ident)),
                                                    to_string(explain())));
        }
        ready_latch.notify_ready();
    }
    spdlog::trace("{}::{} exit", ident, __func__);
}
catch (...)
{
    spdlog::error("{}::{} fatal error {}", ident, __func__, explain());
    condition.reset_error(
        trade::error_info(to_string(to_string(ident)), to_string(explain())));
    ready_latch.notify_ready();
    throw;
}