start.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_op
 * @copyright 2022 ESO - European Southern Observatory
 *
 * @brief Contains definition for the StartAsync operation
 */
#include <daq/op/start.hpp>
 
#include <Metadaqif.hpp>
#include <fmt/format.h>
#include <fmt/ostream.h>
#include <log4cplus/loggingmacros.h>
#include <boost/assert.hpp>
 
#include <daq/op/util.hpp>
#include <daq/state.hpp>
 
namespace {
class RecPropertiesImpl : public recif::RecProperties {
public:
    double getAbsTime() const override {
        return 0.0;
    }
    void setAbsTime(double abs_time) override {
    }
 
    std::string getId() const override {
        return m_id;
    }
    void setId(const std::string& id) override {
        m_id = id;
    }
 
    std::string getInfo() const override {
        return "";
    }
    void setInfo(const std::string& info) override {
    }
 
    std::vector<std::string> getPublishers() const override {
        return {};
    }
    void setPublishers(const std::vector<std::string>& publishers) override {
    }
    bool hasKey() const override {
        return false;
    }
    std::unique_ptr<recif::RecProperties> cloneKey() const override {
        throw std::runtime_error("not clonable");
    }
    std::unique_ptr<recif::RecProperties> clone() const override {
        throw std::runtime_error("not clonable");
    }
    bool keyEquals(const recif::RecProperties& other) const override {
        return false;
    }
 
private:
    std::string m_id;
};
}  // namespace
 
namespace daq::op {
 
StartAsync::StartAsync(AsyncOpParams params_arg) noexcept : m_params(params_arg) {
}
 
boost::future<void> StartAsync::Initiate() {
    // Start metadata then primary
    return StartMeta()
        .then(m_params.executor,
              [this](boost::future<void>&& f) -> boost::future<void> {
                  // Simply propagate errors, if any
                  if (f.has_exception()) {
                      LOG4CPLUS_INFO(m_params.logger,
                                     fmt::format("{}: StartAsync: StartAsync metadaq "
                                                 "failed. Will not start primary data acquisition.",
                                                 m_params.status));
                      // Note: Can't throw directly here as we'll unwrap which then discards that
                      // exception
                      return std::move(f);
                  }
                  // If metadaq was started ok, start primary
                  return StartPrim();
              })
        .unwrap();  // unwrap outer future since we actually want the inner one.
}
 
boost::future<void> StartAsync::StartMeta() {
    return SendRequestAndCollectReplies<void>(
               m_params.meta_sources.begin(),
               m_params.meta_sources.end(),
               // predicate
               [](auto&) { return true; },
               m_params,
               // Sender
               [id = m_params.id](Source<MetaSource>& s) {
                   s.SetState(State::Starting, false);
                   return s.GetSource().GetRrClient().StartDaq(id);
               },
               // reply handler:
               // We transform the return type to void here since there's nothing to keep
               // in the reply
               [](AsyncOpParams params,
                  Source<MetaSource>& source,
                  boost::future<std::shared_ptr<metadaqif::DaqReply>>&& fut) -> void {
                   HandleMetaDaqReply<std::shared_ptr<metadaqif::DaqReply>>("StartDaq",
                                                                            State::Starting,
                                                                            State::Acquiring,
                                                                            State::NotStarted,
                                                                            params,
                                                                            source,
                                                                            std::move(fut));
               },
               std::string_view("StartDaq() (start metadata acquisition)"))
        .then(UnwrapVoidReplies);  // @note: UnwrapVoidReplies is thread safe
}
 
boost::future<void> StartAsync::StartPrim() {
    return SendRequestAndCollectReplies<void>(
               m_params.prim_sources.begin(),
               m_params.prim_sources.end(),
               [](auto&) { return true; },
               m_params,
               // sender
               [id = m_params.id](Source<PrimSource>& s) {
                    auto& client = s.GetSource().GetRrClient();
#if defined(UNIT_TEST)
                   // ECII-783: MAL suddenly requires that interface *must* be implemented by
                   // matching middleware otherwise it fails.
                   // To avoid complicated mocking we still use our local implementation for unit
                   // testing.
                   auto properties = std::make_shared<RecPropertiesImpl>();
#else
 
                   auto mal = client.getMal();
                   BOOST_ASSERT_MSG(mal, "MAL RR client returned invalid MAL pointer");
                   auto properties = mal->createDataEntity<recif::RecProperties>();
                   // Initialize all members by assigning from local implementation
                   *properties = RecPropertiesImpl();
#endif
                   properties->setId(id);
                   s.SetState(State::Starting, false);
                   return client.RecStart(properties);
               },
               // reply handler
               [](AsyncOpParams params,
                  Source<PrimSource>& source,
                  boost::future<std::shared_ptr<recif::RecStatus>>&& fut) -> void {
                   HandlePrimDaqReply<std::shared_ptr<recif::RecStatus>>("RecStart",
                                                                         State::Starting,
                                                                         State::Acquiring,
                                                                         State::NotStarted,
                                                                         params,
                                                                         source,
                                                                         std::move(fut));
               },
               std::string_view("RecStart() (start primary data acquisition)"))
        .then(UnwrapVoidReplies);  // @note: UnwrapVoidReplies is thread safe
}
 
}  // namespace daq::op