dpmClient.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_ocm_libdaq_test
 * @copyright 2022 ESO - European Southern Observatory
 *
 * @brief `daq::DpmClient`
 */
#include <daq/dpmClient.hpp>
 
#include <Daqif.hpp>
#include <mal/Mal.hpp>
#include <mal/rr/qos/ReplyTime.hpp>
 
#include <daqif/subscription.hpp>
#include <daq/conversion.hpp>
 
namespace daq {
 
using DaqReplyPtr = std::shared_ptr<daqif::DaqReply>;
 
struct DpmClientImpl::Subscriptions {
    std::shared_ptr<bool> alive = std::make_shared<bool>(true);
    daqif::Subscription<daqif::DaqStatus> status;
    daqif::Subscription<daqif::StorageStatus> storage;
};
 
DpmClientImpl::DpmClientImpl(boost::asio::io_context& io_ctx, mal::Mal& mal, DpmClientParams params)
    : m_io_ctx(io_ctx), m_executor(m_io_ctx), m_mal(mal), m_params(std::move(params)) {
    m_dpmif = m_mal.getClient<daqif::DpmDaqControlAsync>(
        daqif::MakeServiceUri(m_params.rr_uri, "daq"),
        {std::make_shared<elt::mal::rr::qos::ReplyTime>(params.timeout)},
        {});
    assert(m_dpmif);
 
    m_subs = std::make_unique<Subscriptions>();
    // Add subscription to status notifications, which will invoked from unknown threads,
    // so we dispatch to asio instead and emit signal safely there.
    m_subs->status = daqif::MakeSubscription<daqif::DaqStatus>(
        m_mal,
        daqif::MakeServiceUri(m_params.ps_uri, "daq/status"),
        [weak = std::weak_ptr<bool>(m_subs->alive), this](auto& sub, auto const& event) {
            auto shared = weak.lock();
            if (!shared) {
                // Abandoned
                return;
            }
            auto const& sample = event.getData();
            Status status(sample->getId(), sample->getFileId());
            status << *sample;
            // Dispatch to main thread
            m_io_ctx.post([status = std::move(status), weak, this] {
                auto shared = weak.lock();
                if (!shared) {
                    // Abandoned
                    return;
                }
                m_status_signal(status);
            });
        });
    // Add subscription to storage notifications, which will invoked from unknown threads,
    // so we dispatch to asio instead and emit signal safely there.
    m_subs->storage = daqif::MakeSubscription<daqif::StorageStatus>(
        m_mal,
        daqif::MakeServiceUri(m_params.ps_uri, "dpm/storage"),
        [weak = std::weak_ptr<bool>(m_subs->alive), this](auto& sub, auto const& event) {
            auto shared = weak.lock();
            if (!shared) {
                // Abandoned
                return;
            }
            auto const& sample = event.getData();
            std::filesystem::space_info space = {};
            space << *sample;
            // Dispatch to main thread
            m_io_ctx.post([space, weak, this] {
                auto shared = weak.lock();
                if (!shared) {
                    // Abandoned
                    return;
                }
                m_storage_signal(space);
            });
        });
}
 
DpmClientImpl::~DpmClientImpl() = default;
 
auto DpmClientImpl::ScheduleAsync(std::string const& spec) -> boost::future<State> {
    return m_dpmif->QueueDaq(spec).then(m_executor, [&](boost::future<DaqReplyPtr> f) {
        auto reply = f.get();
        // If it doesn't throw it was successful, until status change is published we can assume
        // state is Scheduled.
        return State::Scheduled;
    });
}
 
auto DpmClientImpl::AbortAsync(std::string const& id) -> boost::future<State> {
    return m_dpmif->AbortDaq(id).then(m_executor, [&](boost::future<DaqReplyPtr> f) {
        auto reply = f.get();
        // If it doesn't throw it was successful, until status change is published we can assume
        // state is Aborted.
        return State::Aborted;
    });
}
 
auto DpmClientImpl::ConnectStatusSignal(StatusSignal::slot_type const& slot)
    -> boost::signals2::connection {
    return m_status_signal.connect(slot);
}
 
auto DpmClientImpl::ConnectStorageSignal(StorageSignal::slot_type const& slot)
    -> boost::signals2::connection {
    return m_storage_signal.connect(slot);
}
 
}  // namespace daq