stop.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_op
 * @copyright 2022 ESO - European Southern Observatory
 *
 * @brief Contains definition for the StopAsync operation
 */
#include <daq/op/stop.hpp>
 
#include <Metadaqif.hpp>
#include <fmt/format.h>
#include <fmt/ostream.h>
#include <log4cplus/loggingmacros.h>
 
#include <daq/fits/json.hpp>
#include <daq/op/util.hpp>
#include <daq/state.hpp>
 
namespace daq::op {
 
StopAsync::StopAsync(ErrorPolicy policy, AsyncOpParams m_params) noexcept
    : m_policy(policy), m_params(m_params), m_error(false) {
}
 
[[nodiscard]] boost::future<Result<DpParts>> StopAsync::Initiate() {
    using boost::future;
    using boost::make_exceptional_future;
 
    // stop primary then metadata.
    return StopPrim()
        .then(m_params.executor,
              [this](future<void> prim_result) -> future<void> {
                  // Simply propagate errors, if any
                  if (prim_result.has_exception()) {
                      m_error = true;
                      if (m_policy == ErrorPolicy::Strict) {
                          LOG4CPLUS_INFO(m_params.logger,
                                         fmt::format("{}: StopAsync: primary daq "
                                                     "failed. Will not stop metadata acquisition.",
                                                     m_params.status));
                          // Note: Can't throw directly here as we'll unwrap which then
                          // discards that exception -> simply propagate future containing
                          // error instead.
                          return make_exceptional_future<void>(prim_result.get_exception_ptr());
                      }
                      LOG4CPLUS_INFO(m_params.logger,
                                     fmt::format("{}: StopAsync: primary daq "
                                                 "failed. Ignoring this because of "
                                                 "ErrorPolicy::Tolerant.",
                                                 m_params.status));
                  }
                  return StopMeta();
              })
        .unwrap()  // transform future<future<T>> to future<T>
        .then([this](future<void> res) -> Result<DpParts> {
            if (res.has_exception()) {
                m_error = true;
                if (m_policy == ErrorPolicy::Strict) {
                    LOG4CPLUS_INFO(m_params.logger,
                                   fmt::format("{}: StopAsync: stopping failed", m_params.status));
                    (void)res.get();  // throws
                }
                LOG4CPLUS_INFO(m_params.logger,
                               fmt::format("{}: StopAsync: meta daq "
                                           "failed. Ignoring this because of "
                                           "ErrorPolicy::Tolerant.",
                                           m_params.status));
            }
            return {m_error, std::move(m_parts)};
        });
}
 
boost::future<void> StopAsync::StopMeta() {
    return SendRequestAndCollectReplies<void>(
               m_params.meta_sources.begin(),
               m_params.meta_sources.end(),
               [](auto&) { return true; },
               m_params,
               // Sender
               [id = m_params.id](Source<MetaSource>& s) {
                   s.SetState(State::Stopping);
                   return s.GetSource().GetRrClient().StopDaq(id);
               },
               // reply handler (note that caller must keep this alive until future is set)
               [this](AsyncOpParams params,
                      Source<MetaSource>& source,
                      boost::future<std::shared_ptr<metadaqif::DaqStopReply>>&& fut) -> void {
                   if (source.GetState() == State::Stopped) {
                       LOG4CPLUS_INFO(params.logger,
                                      fmt::format("{}: StopMeta: Source already stopped, ignoring "
                                                  "reply.",
                                                  params.status));
                       return;
                   }
                   auto reply = HandleMetaDaqReply("StopDaq",
                                                   State::Stopping,
                                                   State::Stopped,
                                                   State::Stopping,
                                                   params,
                                                   source,
                                                   std::move(fut));
                   if (!reply.has_value()) {
                       return;
                   }
                   // Add files and keywords from reply:
                   for (auto const& file : (**reply).getFiles()) {
                       m_parts.emplace_back(std::string(source.GetSource().GetName()), file);
                   }
                   std::string keywords = (**reply).getKeywords();
                   if (!keywords.empty()) {
                       // Decode
                       auto keyword_vec = fits::ParseJsonKeywords(keywords.c_str());
                       m_parts.emplace_back(std::string(source.GetSource().GetName()),
                                            std::move(keyword_vec));
                   }
               },
               std::string_view("StopAsync: stop metadata acquisition"))
        .then(UnwrapVoidReplies);
}
 
boost::future<void> StopAsync::StopPrim() {
    return SendRequestAndCollectReplies<void>(
               m_params.prim_sources.begin(),
               m_params.prim_sources.end(),
               [](Source<PrimSource> const& source) -> bool {
                   /* only send to sources that are not already stopped */
                   return IsSubsequentState(State::Stopped, source.GetState());
               },
               m_params,
               // Sender
               [](Source<PrimSource>& s) {
                   // id is not supported by recif
                   s.SetState(State::Stopping);
                   return s.GetSource().GetRrClient().RecStop();
               },
               // reply handler
               [this](AsyncOpParams params,
                      Source<PrimSource>& source,
                      boost::future<std::shared_ptr<recif::RecStatus>>&& fut) -> void {
                   auto reply = HandlePrimDaqReply("RecStop",
                                                   State::Stopping,
                                                   State::Stopped,
                                                   State::Stopping,
                                                   params,
                                                   source,
                                                   std::move(fut));
                   if (!reply.has_value()) {
                       return;
                   }
                   // Add files and keywords from reply:
                   for (auto const& file : (**reply).getDpFiles()) {
                       m_parts.emplace_back(std::string(source.GetSource().GetName()), file);
                   }
               },
               std::string_view("StopAsync: stop primary data acquisition"))
        .then(UnwrapVoidReplies);
}
 
}  // namespace daq::op