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/signals2/signal.hpp>
#include <boost/thread/future.hpp>
#include <rad/ioExecutor.hpp>
 
#include <daq/status.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;
 
    /**
     * 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);
};
 
class DpmClientImpl : public DpmClient {
public:
    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;
 
    /**
     * 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 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:
    struct Subscriptions;
 
    boost::asio::io_context& m_io_ctx;
    rad::IoExecutor m_executor;
 
    mal::Mal& m_mal;
    DpmClientParams m_params;
 
    StatusSignal m_status_signal;
    StorageSignal m_storage_signal;
 
    std::shared_ptr<daqif::DpmDaqControlAsync> m_dpmif;
    std::unique_ptr<Subscriptions> m_subs;
};
 
} // namespace daq
#endif