dpmClient.hpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_ocm_libdaq_test
 * @copyright 2022 ESO - European Southern Observatory
 *
 * @brief `daq::DpmClient`
 */
#ifndef DAQ_DPMCLIENT_HPP
#define DAQ_DPMCLIENT_HPP
#include "config.hpp"
 
#include <filesystem>
 
#include <boost/asio/io_context.hpp>
#include <boost/asio/steady_timer.hpp>
#include <boost/signals2/signal.hpp>
#include <boost/thread/future.hpp>
#include <rad/ioExecutor.hpp>
#include <log4cplus/logger.h>
 
#include <daq/status.hpp>
#include <daq/resourceToken.hpp>
 
namespace daqif {
class DpmDaqControlAsync;
}
 
namespace elt::mal {
class Mal;
}
namespace mal = ::elt::mal;
 
namespace daq {
 
/**
 * Interface to DPM server.
 *
 * Any implementation is expected to also provide a subscriber for DPM status and storage change
 * signals.
 */
class DpmClient {
public:
    using StorageSignal = boost::signals2::signal<void(std::filesystem::space_info)>;
    using StatusSignal = boost::signals2::signal<void(Status)>;
 
    /**
     * Schedule merging.
     */
    virtual auto ScheduleAsync(std::string const& spec) -> boost::future<State> = 0;
 
    /**
     * Abort merging.
     */
    virtual auto AbortAsync(std::string const& id) -> boost::future<State> = 0;
 
    /**
     * Start monitoring DAQ status.
     *
     * Status is requested:
     * - Initially.
     * - Periodically.
     *
     * Changes are reported via the StatusSignal.
     */
    virtual void StartMonitorStatus(std::string const& id) = 0;
 
    /**
     * Stop monitoring DAQ status.
     */
    virtual void StopMonitorStatus(std::string const& id) = 0;
 
    /**
     * Connect slot to status change signal.
     *
     * DpmClient ensures that signals are only emitted from main thread.
     */
    virtual auto
    ConnectStatusSignal(StatusSignal::slot_type const& slot) -> boost::signals2::connection = 0;
 
    /**
     * Connect slot to storage change signal.
     *
     * DpmClient ensures that signals are only emitted from main thread.
     */
    virtual auto
    ConnectStorageSignal(StorageSignal::slot_type const& slot) -> boost::signals2::connection = 0;
};
 
/**
 * Connection parameters for DPM
 */
struct DpmClientParams {
    std::string rr_uri;
    std::string ps_uri;
    std::chrono::seconds timeout = std::chrono::seconds(5);
    std::chrono::seconds status_retry_interval = std::chrono::seconds(15);
};
 
class DpmClientImpl : public DpmClient {
public:
    /**
     * Retry delay when requesting status.
     */
    using StorageSignal = DpmClient::StorageSignal;
    using StatusSignal = DpmClient::StatusSignal;
 
    DpmClientImpl(boost::asio::io_context& io_ctx, mal::Mal& mal, DpmClientParams params);
 
    ~DpmClientImpl();
 
    /**
     * Schedule merging.
     */
    virtual auto ScheduleAsync(std::string const& spec) -> boost::future<State> override;
 
    /**
     * Abort merging.
     */
    virtual auto AbortAsync(std::string const& id) -> boost::future<State> override;
 
    /**
     * Start monitoring DAQ status.
     *
     * Status is requested:
     * - Initially
     * - and rescheduled if reply times out (DPM not running)
     *
     * Changes are reported via the StatusSignal.
     */
    virtual void StartMonitorStatus(std::string const& id) override;
 
    /**
     * Stop monitoring DAQ status.
     */
    virtual void StopMonitorStatus(std::string const& id) override;
 
    /**
     * Connect slot to status change signal.
     *
     * DpmClient ensures that signals are only emitted from main thread.
     *
     * @note Status may come from different sources and no total order is guaranteed. Observers of
     * status changes should make effort to discard/ignore changes that occurred before already
     * recorded status updated.
     */
    virtual auto ConnectStatusSignal(StatusSignal::slot_type const& slot)
        -> boost::signals2::connection override;
 
    /**
     * Connect slot to storage change signal.
     *
     * DpmClient ensures that signals are only emitted from main thread.
     */
    virtual auto ConnectStorageSignal(StorageSignal::slot_type const& slot)
        -> boost::signals2::connection override;
 
private:
    void RequestStatus(std::string const& id);
    void ScheduleRequestStatus(std::string const& id);
    void UpdateStatus(Status const& status);
 
    struct Subscriptions;
 
    boost::asio::io_context& m_io_ctx;
    rad::IoExecutor m_executor;
 
    mal::Mal& m_mal;
    DpmClientParams m_params;
    log4cplus::Logger m_logger;
 
    StatusSignal m_status_signal;
    StorageSignal m_storage_signal;
 
    std::shared_ptr<daqif::DpmDaqControlAsync> m_dpmif;
    std::unique_ptr<Subscriptions> m_subs;
 
 
    std::unordered_map<std::string, boost::asio::steady_timer> m_timers;
 
    std::shared_ptr<bool> m_liveness = std::make_shared<bool>(true);
    /**
     * Limit number of concurrent status requests.
     */
    Resource m_res_status = Resource(5);
};
 
}  // namespace daq
#endif