ocmDaqService.hpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_ocm
 * @copyright 2021 ESO - European Southern Observatory
 *
 * @brief Declaration of OcmDaqService
 */
#ifndef DAQ_OCM_OCM_DAQ_SERVICE_HPP_
#define DAQ_OCM_OCM_DAQ_SERVICE_HPP_
#include <daq/config.hpp>
#include <rad/ioExecutor.hpp>
 
#include <Ocmif.hpp>
 
#include <boost/asio/io_context.hpp>
#include <daq/eventLog.hpp>
#include <daq/eventLogObserver.hpp>
#include <daq/manager.hpp>
#include <daq/daqProperties.hpp>
#include <log4cplus/logger.h>
#include <mal/Mal.hpp>
 
namespace mal = ::elt::mal;
 
struct ParsedSource {
    ParsedSource(std::string name, std::string rr_uri);
    ParsedSource() = default;
    ParsedSource(ParsedSource const&) = default;
    ParsedSource(ParsedSource&&) = default;
    ParsedSource& operator=(ParsedSource const& rhs) = default;
    ParsedSource& operator=(ParsedSource&& rhs) = default;
 
    bool operator==(ParsedSource const& rhs) const;
    std::string name;
    std::string rr_uri;
};
 
std::ostream& operator<<(std::ostream& os, ParsedSource const& s);
 
/**
 * Parse user provided string in the format
 * "<name>@<rr-uri>"
 *
 * @throw std::invalid_argument on errors.
 */
ParsedSource ParseSourceUri(std::string_view s);
 
/**
 * Parse user provided string in the format
 * "<name>@<rr-uri>[ <name>@...]"
 *
 * @throw std::invalid_argument on errors.
 */
std::vector<ParsedSource> ParseSourceUris(std::string_view s);
 
 
/**
 * Parse the JSON properties user provides with StartDaq
 *
 *  {"keywords": KEYWORDS,
 *   "awaitInterval": DURATION}
 *
 * @throws nlohmann::json::exception on parsing errors.
 */
daq::DaqProperties ParseStartDaqProperties(std::string const& properties);
 
/**
 * Implements the MAL interface ocmif::OcmDaq (async version).
 *
 * The server is only safe to use as a shared pointer, as weak pointers to this object is
 * stored in asynchronous continuations, apart from the initiating operation.
 * Provided boost::asio::io_context must outlive object and any pending operations.
 *
 * This object can be deleted while there are operations that have not complete yet.
 * In that case, when the operation finally completes, an exception is sent as reply, even if
 * the requested operation would have succeeded otherwise. The io_context must stay alive however,
 * as it will be used as an executor when asynchonous operation is continued.
 *
 * A note on thread safety:
 *
 * The public (asynchronous) interface will immediately defer the execution to the provided
 * executor, so depending on the executor this can provide thread safety.
 *
 *
 * @ingroup daq_ocm_server
 */
class OcmDaqService : public ocmif::AsyncOcmDaq,
                      public std::enable_shared_from_this<OcmDaqService> {
public:
    static constexpr char const* LOGGER_NAME = "ocm.service.daq";
 
    /**
     *
     * @param io_ctx ASIO io context to use as an executor. This object must outlive OcmDaqService.
     * The io context will be progated to any created `daq::DaqController` instances as well.
     * @param mal The mal instance to use to create instances. This object must outlive
     * OcmDaqService.
     * @param mgr The data acquisition manager to use. This object must outlive OcmDaqService.
     */
    OcmDaqService(boost::asio::io_context& io_ctx, mal::Mal& mal, daq::Manager& mgr, std::string
                  output_path);
    ~OcmDaqService();
 
    boost::future<std::shared_ptr<::ocmif::DaqReply>>
    StartDaq(const std::string& id,
             const std::string& file_prefix,
             const std::string& primary_sources,
             const std::string& metadata_sources,
             const std::string& properties) override;
 
    boost::future<std::shared_ptr<::ocmif::DaqReply>> StopDaq(const std::string& id) override;
 
    boost::future<std::shared_ptr<::ocmif::DaqReply>>
    ForceStopDaq(const std::string& id) override;
 
    boost::future<std::shared_ptr<::ocmif::DaqReply>>
    AbortDaq(const std::string& id) override;
 
    boost::future<std::shared_ptr<::ocmif::DaqReply>>
    ForceAbortDaq(const std::string& id) override;
 
    boost::future<std::shared_ptr<::ocmif::DaqReply>>
    UpdateKeywords(const std::string& id, const std::string& keywords) override;
 
    boost::future<std::shared_ptr<::ocmif::AwaitDaqReply>>
    AwaitDaqState(const std::string& id,
                  ocmif::DaqState state,
                  ocmif::DaqSubState substate,
                  double timeout) override;
 
    boost::future<std::shared_ptr<::ocmif::DaqStatus>>
    GetStatus(const std::string& id) override;
 
    boost::future<std::vector<std::shared_ptr<::ocmif::DaqStatus>>>
    GetActiveList() override;
 
private:
    boost::future<std::shared_ptr<::ocmif::DaqReply>>
    StopDaq(const std::string& id, bool forced);
    boost::future<std::shared_ptr<::ocmif::DaqReply>>
    AbortDaq(const std::string& id, bool forced);
 
    boost::asio::io_context& m_io_ctx;
    rad::IoExecutor m_executor;
    mal::Mal& m_mal;
    daq::Manager& m_mgr;
    /**
     * Directory in which FITS files are written
     */
    std::string m_output_path;
 
    std::shared_ptr<daq::ObservableEventLog> m_event_log;
    boost::signals2::connection m_log_observer_connection;
    /**
     * Observer for m_event_log.
     */
    daq::EventLogObserverLogger m_log_observer;
 
    log4cplus::Logger m_logger;
};
 
#endif  // #ifndef DAQ_OCM_OCM_DAQ_SERVICE_HPP_