manager.hpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_ocm_libdaq
 * @copyright 2021 ESO - European Southern Observatory
 *
 * @brief Declaration of `daq::Manager`
 */
#ifndef OCM_DAQ_MANAGER_HPP_
#define OCM_DAQ_MANAGER_HPP_
#include "config.hpp"
 
#include <string_view>
#include <vector>
 
#include <boost/thread/future.hpp>
#include <log4cplus/logger.h>
#include <rad/ioExecutor.hpp>
 
#include <daq/state.hpp>
#include <daq/status.hpp>
#include <daq/error.hpp>
#include <daq/utility.hpp>
 
 
namespace daq {
 
/**
 * Creates a DAQ id candidate that may or may not be unique.
 *
 * @param instrument_id The instrument ID to use for the id. Only the 5 first characters will be
 * used if the name is longer than 5.
 * @param jitter Jitter is a millisecond component added to the current time. This is meant to be
 * used to find a unique id by adding jitter until a unique id is found.
 *
 * @note The function cannot guarantee unique ID and is the responsibility of the caller.
 * @ingroup daq_ocm_libdaq
 */
std::string MakeDaqIdCandidate(char const* instrument_id, unsigned jitter=0);
 
class DaqController;
class FitsController;
 
/**
 * Observes any status.
 */
class StatusSignal {
public:
    using SignalType = boost::signals2::signal<void(ObservableStatus const&)>;
 
    template <class Observer>
    boost::signals2::connection ConnectObserver(Observer o) {
        return m_signal.connect(std::move(o));
    }
 
    void Signal(ObservableStatus const& status) {
        m_signal(status);
    }
private:
    SignalType m_signal;
};
 
/**
 * Manager owns DaqController and FitsController (active data acquisitions) instances and
 * multiplexes requests to them.
 *
 *
 * Important responsibilities:
 *
 * - Provide interface that allows multiplexing of multiple DaqControllers and FitsControllers
 *  (identified by unique id). This interface is close to what the MAL ICD interface looks like to
 *  control  DaqControllers and FitsControllers.
 * - Monitor primary data sources to stop DaqController if a primary source completes (TODO).
 *
 * FitsController also act as a data source.
 *
 * Out of scope:
 *
 * - Creation of `DaqController` instances.
 * - Creation of `FitsController` instances.
 *
 * @note: It's a template only to facilitate mocking of DaqController, without requiring it to be
 * an interface.
 *
 * @ingroup daq_ocm_libdaq
 */
class Manager {
public:
    virtual ~Manager() {}
    using Signal = boost::signals2::signal<void(ObservableStatus const&)>;
 
    /**
     * Creates a new unique identifier based on the instrumend id and current time.
     *
     * If there is a id collision when using current time a millisecond jitter component
     * is added until a unique id is found.
     *
     * The format is the same as ARCFILE - the file extension: "<OLAS_ID>-2020-08-19T09:33:11.951"
     *
     * @note OLAS_ID is the first 5 characters of instrument id.
     * @note The returned ID is guaranteed to be unique for all IDs known by Manager (there is
     * currently no global repository of previous IDs).
     *
     * @throws std::system_error on failure.
     */
    virtual std::string MakeDaqId() const = 0;
 
    /**
     * Query existing data acquisition by @c id.
     *
     * @returns true if there is a already a DAQ with the same ID.
     * @returns false otherwise.
     */
    virtual bool HaveDaq(std::string_view id) const DAQ_NOEXCEPT = 0;
 
    /**
     * Get status
     *
     * @throw std::invalid_argument if no data acquisition exist with provided `id`.
     */
    virtual Status GetStatus(std::string_view id) const = 0;
 
    /**
     * Add data acquisition.
     *
     * When a DataAcqusition is added the Manager will also start subscribing to events
     * from primary data sources for monitoring purposes.
     */
    virtual void AddDaq(std::shared_ptr<DaqController> daq) = 0;
 
    /**
     * Start DaqController identified by `id`.
     *
     * @param id Data acquisition id.
     *
     * @returns Exceptional future containing std::invalid_argument, if data acquisition with `id`
     * does not exist.
     * @returns Future that will eventually be ready when data acquisition has started, or failed to
     * start.
     */
    virtual boost::future<State> StartDaqAsync(std::string_view id) = 0;
 
    /**
     * Stop DaqController identified by `id`.
     *
     * @param id Data acquisition id.
     * @param policy Error policy determining if errors are tolerated or not.
     *
     * @returns Exceptional future containing std::invalid_argument, if data acquisition with `id`
     * does not exist.
     * @returns Future that will eventually be ready when data acquisition has stopped, or failed to
     * stop.
     */
    virtual boost::future<Status> StopDaqAsync(std::string_view id, ErrorPolicy policy) = 0;
 
    /**
     * Abort DaqController identified by `id`.
     *
     * @param id Data acquisition id.
     * @param policy Error policy determining if errors are tolerated or not.
     *
     * @returns Exceptional future containing std::invalid_argument, if data acquisition with `id`
     * does not exist.
     * @returns Future that will eventually be ready when data acquisition has aborted, or failed to
     * abort.
     */
    virtual boost::future<Status> AbortDaqAsync(std::string_view id, ErrorPolicy policy) = 0;
 
    /**
     * Await DAQ state
     *
     * @param id Data acquisition id.
     * @param state target state to await.
     * @param timeout How long to wait for state to be reached.
     * @returns Exceptional future containing std::invalid_argument, if data acquisition with `id`
     * does not exist or timeout is negative.
     * @returns Future with a value set when condition is fulfilled or times out.
     */
    virtual boost::future<Result<Status>> AwaitDaqStateAsync(std::string_view id,
                                                             State state,
                                                             std::chrono::milliseconds timeout) = 0;
    /**
     * Update FITS keywords for DaqController identified by `id`.
     * @param id Data acquisition id.
     * @param keywords FITS keywords to update.
     *
     * @throw std::invalid_argument if no data acquisition exist with provided `id`.
     * @throw std::runtime_error If DaqController state does not allow updating of keywords (it's
     * e.g. already been submitted for merging).
     */
    virtual void UpdateKeywords(std::string_view id, fits::KeywordVector const& keywords) = 0;
 
    /**
     * @returns status observer object.
     *
     * The status observer observes any DAQ status changes.
     */
    virtual StatusSignal& GetStatusSignal() = 0;
 
    /**
     * @returns current Daq Controllers.
     */
    virtual std::vector<std::shared_ptr<DaqController const>> GetDaqControllers() = 0;
 
};
 
/**
 * Implements `daq::Manager`.
 *
 * @ingroup daq_ocm_libdaq
 */
class ManagerImpl : public Manager {
public:
    /**
     * @param instrument_id Instrument id.
     */
    explicit ManagerImpl(rad::IoExecutor& executor, std::string instrument_id);
    ~ManagerImpl() noexcept;
 
    std::string MakeDaqId() const override;
 
    bool HaveDaq(std::string_view id) const noexcept override;
 
    Status GetStatus(std::string_view id) const override;
 
    void AddDaq(std::shared_ptr<DaqController> daq) override;
 
    boost::future<State> StartDaqAsync(std::string_view id) override;
 
    boost::future<Status> StopDaqAsync(std::string_view id, ErrorPolicy policy) override;
 
    boost::future<Status> AbortDaqAsync(std::string_view id, ErrorPolicy policy) override;
 
    boost::future<Result<Status>>
    AwaitDaqStateAsync(std::string_view id, State, std::chrono::milliseconds timeout) override;
 
    void UpdateKeywords(std::string_view id, fits::KeywordVector const& keywords) override;
 
    StatusSignal& GetStatusSignal() override;
 
    std::vector<std::shared_ptr<DaqController const>> GetDaqControllers() override;
 
private:
    struct Daq {
        Daq(std::string id_arg,
            std::shared_ptr<DaqController> controller_arg,
            boost::signals2::connection connection_arg) noexcept;
 
        std::string id;
        std::shared_ptr<DaqController> controller;
        // connection for observer connected to controller.
        boost::signals2::connection connection;
    };
 
    /// Adds unique identifier (unrelated to DAQ id) to op abort functions
    class OpAbortFunc {
    public:
        using Func = std::function<bool()>;
        OpAbortFunc(Func&& f);
        OpAbortFunc(OpAbortFunc&&) = default;
        OpAbortFunc& operator=(OpAbortFunc&&) = default;
 
        uint64_t GetId() const noexcept;
        bool Abort() noexcept;
    private:
        static uint64_t NextId();
        uint64_t m_id;
        std::function<bool()> m_func;
    };
    void RemoveAbortFunc(uint64_t id) noexcept;
    DaqController const* FindDaq(std::string_view id) const noexcept;
    DaqController* FindDaq(std::string_view id) noexcept;
    DaqController& FindDaqOrThrow(std::string_view id);
    DaqController const& FindDaqOrThrow(std::string_view id) const;
 
    std::shared_ptr<bool> m_alive_token;
    rad::IoExecutor& m_executor;
    std::string m_instrument_id;
    StatusSignal m_status_signal;
    /// Data acquisitions
    std::vector<Daq> m_daq_controllers;
    /// Pair of DAQ-id and abort function
    std::vector<OpAbortFunc> m_abort_funcs;
    log4cplus::Logger m_logger;
};
 
}  // namespace daq
#endif  // #ifndef OCM_DAQ_MANAGER_HPP_