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/daqContext.hpp>
#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_ERROR(m_params.logger,
                                    fmt::format("{}: StopAsync: stopping failed", m_params.status));
                    (void)res.get();  // throws
                }
                LOG4CPLUS_ERROR(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& source) {
                   // Only send requests to sources that are not stopped.
                   return IsSubsequentState(State::Stopped, source.GetState());
               },
               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,
                       std::nullopt,
                       params,
                       source,
                       std::move(fut),
                       std::function(
                           [&](std::shared_ptr<metadaqif::DaqStopReply> const& rep) -> void {
                               // Add files and keywords from reply. As reply was received we set
                               // source state to Stopped to prevent possiblity of sending request
                               // again.
                               source.SetState(State::Stopped);
                               std::string keywords = rep->getKeywords();
                               if (!keywords.empty()) {
                                   // Decode & validate
                                   fits::KeywordVector keyword_vec;
                                   UpdateKeywords(keyword_vec,
                                                  fits::ParseJsonKeywords(keywords.c_str()),
                                                  m_params.kw_formatter);
                                   m_parts.emplace_back(std::string(source.GetSource().GetName()),
                                                        std::move(keyword_vec));
                               }
                               for (auto const& file : rep->getFiles()) {
                                   m_parts.emplace_back(std::string(source.GetSource().GetName()),
                                                        file);
                               }
                           }));
                   if (!reply.has_value()) {
                       return;
                   }
               },
               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