awaitState.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_common_libdaq
 * @copyright 2022 ESO - European Southern Observatory
 *
 * @brief Contains definitions for the AwaitStateAsync operation
 */
#include <daq/op/awaitState.hpp>
 
#include <fmt/format.h>
#include <fmt/ostream.h>
#include <log4cplus/loggingmacros.h>
 
#include <daq/error.hpp>
 
namespace daq::op {
 
AwaitStateAsync::AwaitStateAsync(boost::asio::io_context& io_ctx,
                                 std::shared_ptr<ObservableStatus> status,
                                 State state,
                                 std::chrono::milliseconds timeout,
                                 log4cplus::Logger const& logger)
    : m_io_ctx(io_ctx)
    , m_status(std::move(status))
    , m_target(state)
    , m_timeout(timeout)
    , m_aborted(false)
    , m_timer(m_io_ctx)
    , m_logger(logger) {
    assert(m_status);
}
 
boost::future<AwaitStateAsync::ResultType> AwaitStateAsync::Initiate() {
    LOG4CPLUS_DEBUG(m_logger,
                    fmt::format("AwaitStateAsync::Initate(): Awaiting state {} with timeout {}ms",
                                fmt::streamed(m_target),
                                m_timeout.count()));
    if (IsConditionSatisfied()) {
        // Condition already satisfied
        return boost::make_ready_future<ResultType>(ResultType{false, m_status->GetStatus()});
    }
 
    // Set up status observer and timer
    m_connection = m_status->ConnectObserver([this](ObservableStatus const&) {
        if (IsConditionSatisfied()) {
            LOG4CPLUS_DEBUG(m_logger,
                            fmt::format("Await condition is satisfied by status change "
                                        "-> setting promise: ",
                                        fmt::streamed(*m_status)));
            try {
                m_promise.set_value(ResultType{false, m_status->GetStatus()});
                m_timer.cancel();
            } catch (boost::promise_already_satisfied const&) {
                // Do nothing.
                LOG4CPLUS_DEBUG(m_logger, "Promise already satisfied");
            }
        }
    });
 
    m_timer.expires_after(m_timeout);
    m_timer.async_wait([this, logger = m_logger](boost::system::error_code const& error) {
        if (error) {
            // Timer was cancelled. This may mean the operation is no longer valid.
            LOG4CPLUS_DEBUG(logger, "Timer cancelled: Doing nothing.");
            return;
        }
        if (m_aborted) {
            // Operation has been aborted. This may mean the operation is no longer valid.
            LOG4CPLUS_INFO(logger, "Operation aborted: Doing nothing.");
            return;
        }
        try {
            // It is only considered a timeout if condition is still not satisfied
            bool timeout = !IsConditionSatisfied();
            LOG4CPLUS_DEBUG(
                m_logger,
                fmt::format("AwaitStateAsync: {}: Timer expired waiting for target state '{}'. "
                            "Current state is '{}'",
                            m_status->GetId(),
                            fmt::streamed(m_target),
                            fmt::streamed(m_status->GetState())));
            m_promise.set_value(ResultType{timeout, m_status->GetStatus()});
        } catch (boost::promise_already_satisfied const&) {
            // May happen due to race-to-complete
            LOG4CPLUS_DEBUG(m_logger, "Promise already satisfied");
        }
    });
 
    return m_promise.get_future();
}
 
void AwaitStateAsync::Abort() noexcept {
    try {
        LOG4CPLUS_INFO(m_logger, "Aborting operation.");
        m_promise.set_exception(DaqOperationAborted(
            fmt::format("AwaitStateAsync: {}: Operation aborted while waiting for target state {}. "
                        "Current state is {}",
                        m_status->GetId(),
                        fmt::streamed(m_target),
                        fmt::streamed(m_status->GetState()))));
        m_aborted = true;
        // Might as well cancel timer and disconnect observer
        m_timer.cancel();
        m_connection.disconnect();
    } catch (boost::promise_already_satisfied const&) {
    }
}
 
bool AwaitStateAsync::IsConditionSatisfied() const {
    return !IsSubsequentState(m_target, m_status->GetState());
}
 
}  // namespace daq::op