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 <cassert>
 
#include <gsl/gsl_util>
 
#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);
 
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);
 
// 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) {
    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 _ = gsl::finally([&] {
        if (!std::uncaught_exceptions()) {
            return;
        }
 
        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 (metadaqif::DaqException const& e) {
        LOG4CPLUS_WARN(params.logger,
                       fmt::format("{}: Metadata source '{}' replied with "
                                   "id='{}'"
                                   "message='{}'",
                                   params.status,
                                   source,
                                   e.getId(),
                                   e.getMessage()));
        params.alerts.Set(
            MakeAlert(alert_id,
                      fmt::format("Metadata source '{}' request '{}' replied "
                                  "with ICD error: {}",
                                  source.GetSource().GetName(),
                                  request,
                                  e.getMessage())));
 
        if (e.getId() != params.id) {
            // Received reply for wrong data acquisition id!
        }
        // source reported error
        throw boost::enable_current_exception(
            DaqSourceError(request, std::string(source.GetSource().GetName()), e.getMessage()));
    } 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)));
        // 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 _ = gsl::finally([&] {
        if (!std::uncaught_exceptions()) {
            return;
        }
 
        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())));
        // 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)));
        // source reported error
        throw boost::enable_current_exception(
            DaqSourceError(request, std::string(source.GetSource().GetName()), what));
    }
}
 
}  // namespace daq::op