util.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_op
 * @copyright 2022 ESO - European Southern Observatory
 *
 * @brief Contains definition for the async op utilities
 */
#include <daq/op/util.hpp>
 
#include <daq/error/report.hpp>
 
namespace daq::op {
 
using metadaqif::DaqReply;
using metadaqif::DaqStopReply;
using recif::RecStatus;
 
template std::optional<std::shared_ptr<DaqReply>> HandleMetaDaqReply<std::shared_ptr<DaqReply>>(
    char const* request,
    std::optional<State> expected_state,
    State success_state,
    std::optional<State> error_state,
    AsyncOpParams params,
    Source<MetaSource>& source,
    boost::future<std::shared_ptr<DaqReply>>&& fut,
    std::function<void(std::shared_ptr<DaqReply> const&)>);
 
template std::optional<std::shared_ptr<DaqStopReply>>
HandleMetaDaqReply<std::shared_ptr<DaqStopReply>>(
    char const* request,
    std::optional<State> expected_state,
    State success_state,
    std::optional<State> error_state,
    AsyncOpParams params,
    Source<MetaSource>& source,
    boost::future<std::shared_ptr<DaqStopReply>>&& fut,
    std::function<void(std::shared_ptr<DaqStopReply> const&)>);
 
// RecAbort return std::string
template std::optional<std::string>
HandlePrimDaqReply<std::string>(char const* request,
                                std::optional<State> expected_state,
                                State success_state,
                                std::optional<State> error_state,
                                AsyncOpParams params,
                                Source<PrimSource>& source,
                                boost::future<std::string>&& fut);
 
template std::optional<std::shared_ptr<RecStatus>>
HandlePrimDaqReply<std::shared_ptr<RecStatus>>(char const* request,
                                               std::optional<State> expected_state,
                                               State success_state,
                                               std::optional<State> error_state,
                                               AsyncOpParams params,
                                               Source<PrimSource>& source,
                                               boost::future<std::shared_ptr<RecStatus>>&& fut);
 
void UnwrapVoidReplies(boost::future<std::vector<boost::future<void>>> futures) {
    LOG4CPLUS_DEBUG("daq", "UnwrapVoidReplies: Entered.");
    std::vector<boost::future<void>> values = futures.get();
    auto exceptions = ExtractExceptions<void>(values);
    if (!exceptions.empty()) {
        LOG4CPLUS_DEBUG("daq", "UnwrapVoidReplies: Throwing gathered exceptions.");
        throw boost::enable_current_exception(DaqSourceErrors(std::move(exceptions)));
    }
    LOG4CPLUS_DEBUG("daq", "UnwrapVoidReplies: Done.");
}
 
template <class ReplyType>
std::optional<ReplyType> HandleMetaDaqReply(char const* request,
                                            std::optional<State> expected_state,
                                            State success_state,
                                            std::optional<State> error_state,
                                            AsyncOpParams params,
                                            Source<MetaSource>& source,
                                            boost::future<ReplyType>&& fut,
                                            std::function<void(ReplyType const&)> func) {
    if (source.GetState() == success_state) {
        // Condition already satisfied.
        LOG4CPLUS_INFO(params.logger,
                       fmt::format("{}: State of source '{}' is already satisfied. "
                                   "Will ignore reply.",
                                   params.status,
                                   source));
        return std::nullopt;
    }
    if (expected_state) {
        if (source.GetState() != *expected_state) {
            LOG4CPLUS_WARN(params.logger,
                           fmt::format("{}: State of source '{}' is in an unexpected state. "
                                       "Expected '{}'. Proceeding normally anyway.",
                                       params.status,
                                       source,
                                       *expected_state));
        }
    }
 
    auto set_error_state = [&] {
        if (error_state) {
            source.SetState(*error_state, true);
        } else {
            source.SetErrorFlag();
        }
    };
 
    auto alert_id =
        MakeAlertId(alert::REQUEST, std::string(request) + source.GetSource().GetName());
 
    try {
        ReplyType r = fut.get();
        if (func) {
            func(r);
        }
        // Success
        source.SetState(success_state, false);
        params.alerts.Clear(alert_id);
        return r;
    } catch (metadaqif::DaqException const& e) {
        auto msg = fmt::format(
            "{}: Metadata source '{}' replied with ICD exception: "
            "id='{}'"
            "message='{}'",
            params.status,
            source,
            e.getId(),
            e.getMessage());
        LOG4CPLUS_WARN(params.logger, msg);
        params.alerts.Set(MakeAlert(alert_id, msg));
        set_error_state();
        // source reported error
        throw boost::enable_current_exception(
            DaqSourceError(request, std::string(source.GetSource().GetName()), e.getMessage()));
    } catch (std::exception const& e) {
        auto msg = fmt::format(
            "{}: Metadata source '{}' reply contains error for id='{}': {}",
            params.status,
            params.id,
            source,
            e.what());
        LOG4CPLUS_WARN(params.logger, msg);
        params.alerts.Set(MakeAlert(alert_id, msg));
        set_error_state();
        // source reported error
        throw boost::enable_current_exception(
            DaqSourceError(request, std::string(source.GetSource().GetName()), e.what()));
    } catch (...) {
        auto what = error::FormatException(std::current_exception());
        LOG4CPLUS_WARN(
            params.logger,
            fmt::format(
                "{}: Metadata source '{}' reply contains error:\n{}", params.status, source, what));
        params.alerts.Set(MakeAlert(alert_id,
                                    fmt::format("Metadata source '{}' request '{}' failed: {}",
                                                source.GetSource().GetName(),
                                                request,
                                                what)));
        set_error_state();
        // source reported error
        throw boost::enable_current_exception(
            DaqSourceError(request, std::string(source.GetSource().GetName()), what));
    }
}
 
template <class ReplyType>
std::optional<ReplyType> HandlePrimDaqReply(char const* request,
                                            std::optional<State> expected_state,
                                            State success_state,
                                            std::optional<State> error_state,
                                            AsyncOpParams params,
                                            Source<PrimSource>& source,
                                            boost::future<ReplyType>&& fut) {
    if (source.GetState() == success_state) {
        // Condition already satisfied.
        // Since multiple concurrent requests are possible, it may be that another request completed
        // the operation.
        LOG4CPLUS_INFO(params.logger,
                       fmt::format("{}: State of source '{}' is already satisfied. "
                                   "Will ignore reply.",
                                   params.status,
                                   source));
        return std::nullopt;
    }
 
    if (expected_state) {
        if (source.GetState() != *expected_state) {
            // @todo It is not likely that we want to proceed normally here. If
            // the expected state changed from e.g. Stopping to Aborting, it doesn't make sense to
            // continue. At the same time, if reply is not handled it might mean data is lost.
            LOG4CPLUS_WARN(params.logger,
                           fmt::format("{}: State of source '{}' is in an unexpected state. "
                                       "Expected '{}'. Proceeding normally anyway.",
                                       params.status,
                                       source,
                                       *expected_state));
        }
    }
 
    auto set_error_state = [&] {
        if (error_state) {
            source.SetState(*error_state, true);
        } else {
            source.SetErrorFlag();
        }
    };
 
    auto alert_id =
        MakeAlertId(alert::REQUEST, std::string(request) + source.GetSource().GetName());
 
    try {
        auto r = fut.get();
        // Success
        source.SetState(success_state, false);
        params.alerts.Clear(alert_id);
        return r;
    } catch (recif::ExceptionErr const& e) {
        LOG4CPLUS_WARN(params.logger,
                       fmt::format("{}: Primdata source '{}' replied with "
                                   "id='{}'"
                                   "message='{}'",
                                   params.status,
                                   source,
                                   e.getCode(),
                                   e.getDesc()));
        params.alerts.Set(MakeAlert(alert_id,
                                    fmt::format("Primary source '{}' request '{}' replied "
                                                "with error: ({}) {}",
                                                source.GetSource().GetName(),
                                                request,
                                                e.getCode(),
                                                e.getDesc())));
        set_error_state();
        // source reported error
        throw boost::enable_current_exception(
            DaqSourceError(request, std::string(source.GetSource().GetName()), e.getDesc()));
    } catch (...) {
        auto what = error::FormatException(std::current_exception());
        LOG4CPLUS_WARN(
            params.logger,
            fmt::format("{}: Primdata source '{}' failed: {}", params.status, source, what));
        params.alerts.Set(MakeAlert(alert_id,
                                    fmt::format("Primary source '{}' request '{}' failed: {}",
                                                source.GetSource().GetName(),
                                                request,
                                                what)));
        set_error_state();
        // source reported error
        throw boost::enable_current_exception(
            DaqSourceError(request, std::string(source.GetSource().GetName()), what));
    }
}
 
}  // namespace daq::op