asyncOpParams.hpp

Go to the documentation of this file.

#ifndef DAQ_OP_ASYNC_OP_PARAMS_HPP_
#define DAQ_OP_ASYNC_OP_PARAMS_HPP_
 
#include <boost/asio/io_context.hpp>
#include <log4cplus/logger.h>
 
#include <rad/ioExecutor.hpp>
 
#include <daq/fits/keyword.hpp>
#include <daq/eventLog.hpp>
#include <daq/pendingReplies.hpp>
#include <daq/source.hpp>
#include <daq/status.hpp>
 
namespace daq::op {
 
struct AlertState {
    std::vector<Alert> set;
    std::vector<Alert> cleared;
 
    void Set(Alert alert) {
        SetAlert(set, std::move(alert));
    }
    void Clear(AlertId id) {
        auto alert = MakeAlert(std::move(id), "");
        auto it = std::find(cleared.begin(), cleared.end(), alert);
        if (it != cleared.end()) {
            // Replace
            *it = std::move(alert);
        } else {
            // Or add
            cleared.emplace_back(std::move(alert));
        }
    }
};
 
/**
 * Merge alerts.
 *
 * It only set/clears if timestamp is newer.
 */
inline void MergeAlerts(ObservableStatus& dest, AlertState& src) {
    auto const& active = dest.GetAlerts();
    // Order should not be important
    for (auto& a : src.set) {
        auto it = std::find(active.begin(), active.end(), a);
        // Only set alert if new or newer
        if (it == active.end() || a.timestamp > it->timestamp) {
            dest.SetAlert(std::move(a));
        }
    }
    for (auto& a : src.cleared) {
        auto it = std::find(active.begin(), active.end(), a);
        // Only clear alert if timestamp of `a` is newer
        if (it != active.end() && a.timestamp > it->timestamp) {
            dest.ClearAlert(a.id);
        }
    }
}
 
/**
 * Parameters required for each async operation.
 *
 * Caller is responsible for keeping objects alive for the duration of the operation.
 */
struct AsyncOpParams {
    AsyncOpParams(ObservableStatus& status_arg,
                  ObservableEventLog& event_log_arg,
                  AlertState& alerts,
                  rad::IoExecutor& executor_arg,
                  log4cplus::Logger& logger_arg,
                  std::string const& id_arg,
                  PendingReplies& pending_replies_arg,
                  std::vector<Source<PrimSource>>& prim_sources_arg,
                  std::vector<Source<MetaSource>>& meta_sources_arg,
                  fits::KeywordFormatter const& kw_formatter_arg)
        : status(status_arg)
        , event_log(event_log_arg)
        , alerts(alerts)
        , executor(executor_arg)
        , logger(logger_arg)
        , id(id_arg)
        , pending_replies(pending_replies_arg)
        , prim_sources(prim_sources_arg)
        , meta_sources(meta_sources_arg)
        , kw_formatter(kw_formatter_arg) {
    }
    AsyncOpParams(AsyncOpParams const&) = default;
    AsyncOpParams(AsyncOpParams&&) = default;
 
    /**
     * Async operations should not modify state directly
     * DaqController does that. The only allowed operation is to set/clear alerts.
     */
    ObservableStatus& status;
    ObservableEventLog& event_log;
    /**
     * Alerts to be merged only after completion of async operation.
     */
    AlertState& alerts;
    rad::IoExecutor& executor;
    log4cplus::Logger& logger;
    std::string const& id;
    PendingReplies& pending_replies;
    std::vector<Source<PrimSource>>& prim_sources;  ///< Note: Consider vector immutable!
    std::vector<Source<MetaSource>>& meta_sources;  ///< Note: Consider vector immutable!a
    fits::KeywordFormatter const& kw_formatter;
};
 
/**
 * Await specific parameters that is not provided with AsyncOpParams.
 *
 * @important The clients provided must be configured with a reply timeout that takes the wait
 * timeout into account, otherwise MAL will time out before reply is sent.
 */
struct AwaitOpParams {
    using Duration = std::chrono::milliseconds;
    AwaitOpParams(AsyncOpParams common_arg, AwaitOpParams::Duration wait_interval_arg) noexcept
        : common(common_arg), wait_interval(wait_interval_arg) {
    }
    AwaitOpParams(AwaitOpParams const&) = default;
    AwaitOpParams(AwaitOpParams&&) = default;
 
    AsyncOpParams common;
 
    /**
     * Total amount of time to wait for condition to be fulfilled.
     */
    Duration wait_interval;
};
 
}  // namespace daq::op
 
#endif  // #ifndef DAQ_OP_ASYNC_OP_PARAMS_HPP_