daqController.cpp

Go to the documentation of this file.

#include <daq/daqController.hpp>
 
#include <iostream>
#include <stdexcept>
#include <type_traits>
 
#include <fmt/format.h>
#include <fmt/ostream.h>
#include <log4cplus/loggingmacros.h>
#include <mal/MalException.hpp>
#include <mal/rr/qos/ReplyTime.hpp>
#include <nlohmann/json.hpp>
 
#include <daq/dpmClient.hpp>
#include <daq/error.hpp>
#include <daq/fits/json.hpp>
#include <daq/op/abort.hpp>
#include <daq/op/awaitPrim.hpp>
#include <daq/op/initiate.hpp>
#include <daq/op/start.hpp>
#include <daq/op/stop.hpp>
 
namespace daq {
namespace {
 
DaqSources MakeSources(mal::Mal& mal, DaqContext const& ctx) {
    std::vector<PrimSource> psources;
    std::vector<MetaSource> msources;
 
    psources.reserve(ctx.prim_sources.size());
    for (auto const& raw : ctx.prim_sources) {
        psources.emplace_back(
            raw.name,
            mal.getClient<daq::PrimSource::RrClient>(
                mal::Uri(raw.rr_uri),
                {std::make_shared<elt::mal::rr::qos::ReplyTime>(std::chrono::seconds(5))},
                {}));
    }
 
    msources.reserve(ctx.meta_sources.size());
    for (auto const& raw : ctx.meta_sources) {
        msources.emplace_back(
            raw.name,
            mal.getClient<daq::MetaSource::RrClient>(
                mal::Uri(raw.rr_uri),
                {std::make_shared<elt::mal::rr::qos::ReplyTime>(std::chrono::seconds(5))},
                {}));
    }
 
    return DaqSources(std::move(psources), std::move(msources));
}
 
template <class>
inline constexpr bool always_false_v = false;  // NOLINT
 
template <class Sources>
bool AllInState(Sources sources, State state) {
    return std::all_of(
        sources.begin(), sources.end(), [=](auto const& s) -> bool { return s.state == state; });
}
 
}  // namespace
 
OcmAsyncOperations::OcmAsyncOperations()
    : start([](auto par) { return daq::op::InitiateOperation<daq::op::StartAsync>(par); })
    , abort([](ErrorPolicy policy, auto par) {
        return daq::op::InitiateOperation<op::AbortAsync>(policy, std::move(par));
    })
    , stop([](ErrorPolicy policy, auto par) {
        return daq::op::InitiateOperation<op::StopAsync>(policy, std::move(par));
    })
    , await_prim([](auto par) {
        return daq::op::InitiateAbortableOperation<op::AwaitPrimAsync>(std::move(par));
    }) {
}
 
bool OcmAsyncOperations::IsValid() const noexcept {
    return start && stop && abort && await_prim;
}
 
DaqControllerFactoryDefault::DaqControllerFactoryDefault(boost::asio::io_context& io_ctx,
                                                         elt::mal::Mal& mal,
                                                         std::shared_ptr<DpmClient> dpm_client)
    : m_io_ctx(io_ctx), m_mal(mal), m_async_ops(), m_dpm_client(std::move(dpm_client)) {
    assert(m_dpm_client);
}
 
auto DaqControllerFactoryDefault::MakeOcmPhase(DaqContext daq_ctx,
                                               std::shared_ptr<ObservableStatus> status,
                                               std::shared_ptr<ObservableEventLog> event_log)
    -> std::shared_ptr<DaqController> {
    DaqSources sources = MakeSources(m_mal, daq_ctx);
    return OcmDaqController::Create(m_io_ctx,
                                    std::move(daq_ctx),
                                    sources,
                                    std::move(status),
                                    std::move(event_log),
                                    m_async_ops);
}
 
auto DaqControllerFactoryDefault::MakeDpmPhase(DaqContext daq_ctx,
                                               std::shared_ptr<ObservableStatus> status,
                                               std::shared_ptr<ObservableEventLog> event_log)
    -> std::shared_ptr<DaqController> {
    return std::make_shared<DpmDaqController>(
        m_io_ctx, std::move(daq_ctx), std::move(status), std::move(event_log), m_dpm_client);
}
 
std::ostream& operator<<(std::ostream& os, DaqController const& daq) {
    os << "DaqController(id='" << daq.GetId() << "', state=" << daq.GetState() << ")";
    return os;
}
 
CommonDaqController::CommonDaqController(boost::asio::io_context& io_context,
                                         DaqContext context,
                                         std::shared_ptr<ObservableStatus> status,
                                         std::shared_ptr<ObservableEventLog> event_log)
    : m_io_ctx(io_context)
    , m_executor(m_io_ctx)
    , m_context(std::move(context))
    , m_status(std::move(status))
    , m_event_log(std::move(event_log)) {
    assert(m_status);
    assert(m_event_log);
}
 
std::shared_ptr<ObservableStatus> CommonDaqController::GetStatus() DAQ_NOEXCEPT {
    return m_status;
}
 
std::shared_ptr<ObservableStatus const> CommonDaqController::GetStatus() const DAQ_NOEXCEPT {
    return m_status;
}
 
std::shared_ptr<ObservableEventLog> CommonDaqController::GetEventLog() DAQ_NOEXCEPT {
    return m_event_log;
}
 
std::string const& CommonDaqController::GetId() const DAQ_NOEXCEPT {
    return m_status->GetId();
}
 
bool CommonDaqController::GetErrorFlag() const DAQ_NOEXCEPT {
    return m_status->GetError();
}
 
DaqContext const& CommonDaqController::GetContext() const DAQ_NOEXCEPT {
    return m_context;
}
 
boost::signals2::connection
CommonDaqController::ConnectContext(ContextSignal::slot_type const& slot) {
    return m_sig_context.connect(slot);
}
 
std::shared_ptr<OcmDaqController>
OcmDaqController::Create(boost::asio::io_context& io_context,
                         DaqContext context,
                         DaqSources const& sources,
                         std::shared_ptr<ObservableStatus> status,
                         std::shared_ptr<ObservableEventLog> event_log,
                         OcmAsyncOperations ops) {
    // note: make_shared doesn't work since constructor is protected,
    // to protect against non-shared ownership.
    LOG4CPLUS_TRACE("daq", fmt::format("OcmDaqController::Create"));
    return std::shared_ptr<OcmDaqController>(new OcmDaqController(io_context,
                                                                  std::move(context),
                                                                  sources,
                                                                  std::move(status),
                                                                  std::move(event_log),
                                                                  std::move(ops)));
}
 
OcmDaqController::OcmDaqController(boost::asio::io_context& io_context,
                                   DaqContext context,
                                   DaqSources const& sources,
                                   std::shared_ptr<ObservableStatus> status,
                                   std::shared_ptr<ObservableEventLog> event_log,
                                   OcmAsyncOperations ops)
    : CommonDaqController(io_context, std::move(context), std::move(status), std::move(event_log))
 
    , m_state(MakeState(GetStatusRef().GetState()))
    , m_prim_sources(MakeSources<PrimSource>(sources.GetPrimarySources()))
    , m_meta_sources(MakeSources<MetaSource>(sources.GetMetadataSources()))
    , m_async_ops(std::move(ops))
    , m_pending_replies(PendingReplies::Create())
    , m_logger(log4cplus::Logger::getInstance("daq")) {
    if (GetContext().id != GetStatusRef().GetId()) {
        throw boost::enable_current_exception(
            std::invalid_argument("Data acquisition id mismatch between DaqContext "
                                  "and ObservableStatus"));
    }
 
    if (m_prim_sources.empty() && m_meta_sources.empty()) {
        throw boost::enable_current_exception(std::invalid_argument("No data sources provided"));
    }
 
    if (!m_async_ops.IsValid()) {
        throw boost::enable_current_exception(
            std::invalid_argument("OcmAsyncOperations is invalid"));
    }
}
 
State OcmDaqController::GetState() const DAQ_NOEXCEPT {
    State s;
    std::visit(
        [&](auto const& var) {
            using T = std::decay_t<decltype(var)>;
            if constexpr (std::is_same_v<T, NotStarted>) {
                s = State::NotStarted;
            } else if constexpr (std::is_same_v<T, Starting>) {
                s = State::Starting;
            } else if constexpr (std::is_same_v<T, Acquiring>) {
                s = State::Acquiring;
            } else if constexpr (std::is_same_v<T, Stopping>) {
                s = State::Stopping;
            } else if constexpr (std::is_same_v<T, Stopped>) {
                s = State::Stopped;
            } else if constexpr (std::is_same_v<T, Aborting>) {
                s = State::AbortingAcquiring;
            } else if constexpr (std::is_same_v<T, Aborted>) {
                s = State::Aborted;
            } else {
                static_assert(always_false_v<T>, "non-exhaustive visitor!");
            }
        },
        m_state);
    return s;
}
 
constexpr log4cplus::Logger const& OcmDaqController::GetLogger() const noexcept {
    return m_logger;
}
 
OcmDaqController::StateVariant OcmDaqController::MakeState(State s) const noexcept {
    switch (s) {
    case State::NotStarted:
        return StateVariant(NotStarted());
    case State::Starting:
        return StateVariant(Starting());
    case State::Acquiring:
        return StateVariant(Acquiring());
    case State::Stopping:
        return StateVariant(Stopping());
    case State::Stopped:
        return StateVariant(Stopped());
    case State::AbortingAcquiring:
        return StateVariant(Aborting());
    case State::Aborted:
        return StateVariant(Aborted());
    default:
        break;
    };
    LOG4CPLUS_FATAL(m_logger, fmt::format("Invalid state provided: '{}'", s));
    std::terminate();
}
 
void OcmDaqController::SetErrorFlag(bool error) noexcept {
    GetStatusRef().SetError(error);
}
 
void OcmDaqController::SetState(OcmDaqController::StateVariant&& s) noexcept {
    m_state = s;
    // Publish changes
    GetStatusRef().SetState(GetState());
}
 
op::AsyncOpParams OcmDaqController::MakeParams(op::AlertState& alerts) {
    return op::AsyncOpParams(GetStatusRef(),
                             GetEventLogRef(),
                             alerts,
                             GetIoExecutor(),
                             m_logger,
                             GetId(),
                             *m_pending_replies.get(),
                             m_prim_sources,
                             m_meta_sources);
}
 
op::AwaitOpParams OcmDaqController::MakeAwaitParams(op::AlertState& alerts) {
    return op::AwaitOpParams(op::AsyncOpParams(GetStatusRef(),
                                               GetEventLogRef(),
                                               alerts,
                                               GetIoExecutor(),
                                               m_logger,
                                               GetId(),
                                               *m_pending_replies.get(),
                                               m_prim_sources,
                                               m_meta_sources),
                             GetContext().await_interval);
}
 
void OcmDaqController::AddInitialKeywords() {
    // Nothing yet.
}
 
boost::future<State> OcmDaqController::StartAsync() {
    GetEventLogRef().AddEvent(
        ActionEvent(GetId(), "DaqController::StartAsync()", GetStatusRef().GetStatus()));
 
    // Make sure we're not already started.
    if (!std::holds_alternative<NotStarted>(m_state)) {
        return boost::make_exceptional_future<State>(
            std::runtime_error("Data acquisition is already started"));
    }
 
    SetState(Starting{});
 
    AddInitialKeywords();
 
    auto alerts = std::make_unique<op::AlertState>();
    return m_async_ops.start(MakeParams(*alerts.get()))
        .then(GetIoExecutor(),
              [alerts = std::move(alerts),
               daq = std::static_pointer_cast<OcmDaqController>(shared_from_this())](
                  boost::future<void> f) {
                  LOG4CPLUS_INFO(daq->GetLogger(),
                                 fmt::format("{}: StartAsync: Completed {}",
                                             *daq,
                                             f.has_value() ? "successfully" : "with error"));
                  // Merge alerts from async op
                  auto defer = ObservableStatus::DeferSignal(&daq->GetStatusRef());
                  op::MergeAlerts(daq->GetStatusRef(), *alerts);
 
                  // Re-raise any exception
                  if (f.has_exception()) {
                      daq->SetErrorFlag(true);
                      (void)f.get();
                  }
                  // Await completion if there are primary data sources
                  daq->InitiateAwaitPrimarySources();
 
                  // No exception -> all done! update and return Acquiring state
                  daq->SetErrorFlag(false);
                  daq->SetState(Acquiring{});
                  return daq->GetState();
              });
}
 
boost::future<Status> OcmDaqController::StopAsync(ErrorPolicy policy) {
    GetEventLogRef().AddEvent(
        ActionEvent(GetId(), "DaqController::StopAsync()", GetStatusRef().GetStatus()));
 
    if (std::holds_alternative<Stopped>(m_state)) {
        return boost::make_exceptional_future<Status>(
            std::runtime_error("Data acquisition already stopped"));
    }
 
    if (std::holds_alternative<Aborted>(m_state)) {
        return boost::make_exceptional_future<Status>(
            std::runtime_error("Data acquisition already aborted"));
    }
 
    if (!std::holds_alternative<Acquiring>(m_state)) {
        return boost::make_exceptional_future<Status>(
            std::runtime_error("Cannot stop a data acquisition that is not Acquiring"));
    }
 
    // @todo If we're in Starting this will potentially mess up assumption that we're
    // in Starting. Revisit to remove this assumption?
    // @todo: Store produced files
    // m_status.AddFiles(reply.getFiles());
    SetState(Stopping{});
 
    auto alerts = std::make_unique<op::AlertState>();
    return m_async_ops.stop(policy, MakeParams(*alerts.get()))
        .then(GetIoExecutor(),
              [alerts = std::move(alerts),
               daq = std::static_pointer_cast<OcmDaqController>(shared_from_this())](
                  boost::future<Result<DpParts>> f) mutable -> Status {
                  if (daq->GetState() != State::Stopping) {
                      // It can happen that a request to stop was superseded and completed before
                      // reply was received. In this case we treat it as a failure as we cannot
                      // determine if post conditions are met.
                      LOG4CPLUS_WARN(
                          daq->GetLogger(),
                          fmt::format("{}: StopAsync: Data acquisition modified by other commands. "
                                      "Do nothing else (errors are ignored). ",
                                      *daq));
                      throw std::runtime_error(fmt::format(
                          "Stop command could not be completed because Data Acquisitions state was "
                          "modified in the meantime (current state: {})",
                          daq->GetState()));
                  }
                  LOG4CPLUS_INFO(daq->GetLogger(),
                                 fmt::format("{}: StopAsync: Completed {}",
                                             *daq,
                                             f.has_value() ? "successfully" : "with error"));
                  auto defer = ObservableStatus::DeferSignal(&daq->GetStatusRef());
                  op::MergeAlerts(daq->GetStatusRef(), *alerts);
 
                  if (f.has_exception()) {
                      daq->SetErrorFlag(true);
                      (void)f.get();  // Throw to propagate any error
                  }
 
                  auto result = f.get();
                  AddDpParts(daq->GetContextMut(), result.result);
                  daq->EmitContextSignal();
 
                  // If there were no exceptions we're all done.
                  // Update and return Stopped state
                  daq->SetErrorFlag(result.error);
                  daq->SetState(Stopped{});
                  return daq->GetStatus()->GetStatus();
              });
}
 
boost::future<Status> OcmDaqController::AbortAsync(ErrorPolicy policy) {
    GetEventLogRef().AddEvent(ActionEvent(
        GetId(), fmt::format("DaqController::AbortAsync({})", policy), GetStatusRef().GetStatus()));
 
    if (std::holds_alternative<NotStarted>(m_state)) {
        LOG4CPLUS_INFO(m_logger, fmt::format("{}: Aborting not started data acquisition", *this));
        SetState(Aborted{});
        return boost::make_ready_future<Status>(GetStatus()->GetStatus());
    }
 
    if (std::holds_alternative<Stopped>(m_state) || std::holds_alternative<Aborted>(m_state)) {
        return boost::make_exceptional_future<Status>(
            std::runtime_error("Data acquisition already stopped or aborted"));
    }
 
    SetState(Aborting{});
 
    auto alerts = std::make_unique<op::AlertState>();
    return m_async_ops.abort(policy, MakeParams(*alerts.get()))
        .then(GetIoExecutor(),
              [alerts = std::move(alerts),
               daq = std::static_pointer_cast<OcmDaqController>(shared_from_this())](
                  boost::future<Result<void>> f) -> Status {
                  if (daq->GetState() == State::Aborted) {
                      // It can happen that a request to abort was superseded and
                      // finalized before reply was received.
                      LOG4CPLUS_INFO(
                          daq->GetLogger(),
                          fmt::format("{}: AbortAsync: Data acquisition already aborted. "
                                      "Do nothing else (errors are ignored). ",
                                      *daq));
                      // @todo: Should throw instead as the command was superseeded by
                      // other command.
                      return daq->GetStatus()->GetStatus();
                  }
                  LOG4CPLUS_DEBUG(daq->GetLogger(),
                                  fmt::format("{}: AbortAsync: Completed, updating DAQ status and "
                                              "set reply remaining. Has fatal error={}",
                                              *daq,
                                              f.has_exception()));
                  //
                  // Merge alerts from saync op
                  auto defer = ObservableStatus::DeferSignal(&daq->GetStatusRef());
                  op::MergeAlerts(daq->GetStatusRef(), *alerts);
 
                  if (f.has_exception()) {
                      LOG4CPLUS_ERROR(
                          daq->GetLogger(),
                          fmt::format("{}: AbortAsync: Completed with fatal error.", *daq));
 
                      daq->SetErrorFlag(true);
                      f.get();  // throw to propagate
                  }
                  auto result = f.get();
                  daq->SetErrorFlag(result.error);
 
                  // Success
                  daq->SetState(Aborted{});
                  LOG4CPLUS_INFO(daq->GetLogger(),
                                 fmt::format("{}: AbortAsync: Completed successfully.", *daq));
                  return daq->GetStatus()->GetStatus();
              });
}
 
boost::future<State> OcmDaqController::ScheduleMergeAsync() {
    return boost::make_exceptional_future<State>(std::runtime_error(
        fmt::format("ScheduleMergeAsync() is invalid in state: {}", GetState())));
}
 
void OcmDaqController::UpdateKeywords(fits::KeywordVector const& keywords) {
    GetEventLogRef().AddEvent(ActionEvent(GetId(),
                                          fmt::format("DaqController::UpdateKeywords(<omitted>)"),
                                          GetStatusRef().GetStatus()));
 
    if (std::holds_alternative<Stopped>(m_state) || std::holds_alternative<Aborted>(m_state)) {
        throw boost::enable_current_exception(
            std::runtime_error("Data acquisition already stopped or aborted"));
    }
 
    daq::UpdateKeywords(GetContextMut(), keywords);
    EmitContextSignal();
}
 
boost::future<State>
OcmDaqController::AwaitAsync(std::vector<std::string> sources, std::chrono::milliseconds timeout) {
    GetEventLogRef().AddEvent(ActionEvent(
        GetId(),
        fmt::format("DaqController::AwaitAsync({}, {} ms)",
                    sources.empty() ? "all primary sources" : "a user defined list of sources",
                    timeout.count()),
        GetStatusRef().GetStatus()));
 
    std::vector<
        std::variant<gsl::not_null<Source<PrimSource>*>, gsl::not_null<Source<MetaSource>*>>>
        await_on;
 
    if (!sources.empty()) {
        for (auto const& source_id : sources) {
            auto source_var = FindSource(source_id);
            if (source_var) {
                await_on.emplace_back(*source_var);
            } else {
                return boost::make_exceptional_future<State>(
                    std::invalid_argument(fmt::format("Source with id='{}' not found", source_id)));
            }
        }
    } else {
        // Use all primary sources by default
        for (auto& source : m_prim_sources) {
            await_on.emplace_back(&source);
        }
    }
 
    // note that condition references sources in OcmDaqController and should not be invoked
    // unless OcmDaqController is alive.
    auto condition = [sources = await_on]() {
        return std::all_of(sources.begin(), sources.end(), [](auto var) {
            return std::visit(
                [](auto v) {
                    return v->GetState() == State::Aborted || v->GetState() == State::Stopped;
                },
                var);
        });
    };
 
    // Test if condition is already satified
    if (condition()) {
        return boost::make_ready_future<State>(GetState());
    }
 
    // Wait is not already satisfied, attach state listeners to all sources.
    // daq is captured with a weak ptr to avoid keeping OcmDaqController alive if no state changes
    // occur on monitored components.
    auto promise = std::make_shared<boost::promise<State>>();
 
    if (timeout > std::chrono::milliseconds(0)) {
        auto timer = std::make_unique<boost::asio::steady_timer>(GetIoCtx(), timeout);
        timer->async_wait([promise,
                           timer_ptr = timer.get(),
                           daq_weak = std::weak_ptr<OcmDaqController>(
                               std::static_pointer_cast<OcmDaqController>(shared_from_this()))](
                              boost::system::error_code ec) {
            // Promise might already be fulfilled, there's no way to check if it is though.
            try {
                auto daq = daq_weak.lock();
                if (ec) {
                    if (daq) {
                        LOG4CPLUS_DEBUG(
                            daq->m_logger,
                            fmt::format("{}: AsyncWait: Operation abandoned before completing.",
                                        *daq));
                    }
                    // Timer deleted or was cancelled, set exception in promise
                    promise->set_exception(DaqOperationAborted(""));
                } else {
                    // Normal timeout
                    // For this case we also want to delete the timer itself
                    if (daq) {
                        LOG4CPLUS_DEBUG(
                            daq->m_logger,
                            fmt::format("{}: AsyncWait: Operation timed out before completing.",
                                        *daq));
                        daq->m_timers.erase(
                            std::remove_if(
                                daq->m_timers.begin(),
                                daq->m_timers.end(),
                                [timer_ptr](std::unique_ptr<boost::asio::steady_timer>& val) {
                                    return val.get() == timer_ptr;
                                }),
                            daq->m_timers.end());
                    }
 
                    promise->set_exception(DaqOperationTimeout(""));
                }
            } catch (...) {
            }
        });
        m_timers.emplace_back(std::move(timer));
    }
 
    auto listener = [condition,
                     promise,
                     daq_weak = std::weak_ptr<OcmDaqController>(
                         std::static_pointer_cast<OcmDaqController>(shared_from_this()))](State,
                                                                                          bool) {
        auto daq = daq_weak.lock();
        if (!daq) {
            LOG4CPLUS_WARN("daq", "OcmDaqController deleted before await condition was fulfulled");
            // this async op was abandoned. Do nothing.
            return;
        }
 
        if (condition()) {
            LOG4CPLUS_INFO(daq->m_logger,
                           fmt::format("{}: AwaitAsync: Await condition fulfilled", *daq));
            try {
                promise->set_value(daq->GetState());
            } catch (...) {
                // Exception might be thrown because promise is already fulfilled, which is
                // expected to happen.
            }
        }
    };
 
    // Connect listeners to sources that should be awaited.
    for (auto& source : await_on) {
        std::visit(
            [daq = std::static_pointer_cast<OcmDaqController>(shared_from_this()),
             listener](auto s) {
                LOG4CPLUS_DEBUG(
                    daq->m_logger,
                    fmt::format("{}: AsyncWait: Attaching listener on source '{}'.", *daq, *s));
                // Use automatic connection disconnect if daq is destroyed. (i.e. track_foreign).
                using SlotType = typename std::remove_reference<
                    decltype(*s.get())>::type::StateSignal::slot_type;
                s->ConnectStateListener(SlotType(std::move(listener)).track_foreign(daq));
            },
            source);
    }
    return promise->get_future();
}
 
std::optional<std::variant<gsl::not_null<Source<PrimSource>*>, gsl::not_null<Source<MetaSource>*>>>
OcmDaqController::FindSource(std::string_view source_id) {
    {
        auto it =
            std::find_if(m_prim_sources.begin(), m_prim_sources.end(), [source_id](auto& source) {
                return source.GetSource().GetName() == source_id;
            });
        if (it != m_prim_sources.end()) {
            return gsl::not_null<Source<PrimSource>*>(&(*it));
        }
    }
    {
        auto it =
            std::find_if(m_meta_sources.begin(), m_meta_sources.end(), [source_id](auto& source) {
                return source.GetSource().GetName() == source_id;
            });
        if (it != m_meta_sources.end()) {
            return gsl::not_null<Source<MetaSource>*>(&(*it));
        }
    }
 
    // Not found
    return {};
}
 
template <class SourceType>
std::vector<Source<SourceType>> OcmDaqController::MakeSources(std::vector<SourceType> sources) {
    //@todo: Check for duplicates
    std::vector<Source<SourceType>> dest;
    dest.reserve(sources.size());
    std::transform(
        std::make_move_iterator(sources.begin()),
        std::make_move_iterator(sources.end()),
        std::back_inserter(dest),
        [](SourceType&& s) -> Source<SourceType> { return Source<SourceType>{std::move(s)}; });
    return dest;
}
 
void OcmDaqController::InitiateAwaitPrimarySources() {
    AddEvent<ActionEvent>(GetId(),
                          "DaqController::InitiateAwaitPrimarySources(): Initiating",
                          GetStatusRef().GetStatus());
    if (m_prim_sources.empty()) {
        LOG4CPLUS_DEBUG(
            m_logger,
            fmt::format("{}: InitiateAwaitPrimarySources: No primary sources to monitor.", *this));
        return;
    }
    LOG4CPLUS_DEBUG(m_logger,
                    fmt::format("{}: InitiateAwaitPrimarySources: Starting operation.", *this));
    auto alerts = std::make_unique<op::AlertState>();
    auto [future, abort] = m_async_ops.await_prim(MakeAwaitParams(*alerts.get()));
    m_abort_await_primary_sources = abort;
    // Set up continuation that stops data acquisition automatically
    future.then(
        GetIoExecutor(),
        [alerts = std::move(alerts),
         daq = std::static_pointer_cast<OcmDaqController>(shared_from_this())](
            boost::future<Result<DpParts>> fut) {
            // Add FITS files from the stopped sources
            auto result = fut.get();
            LOG4CPLUS_DEBUG(daq->m_logger,
                            fmt::format("{}: InitiateAwaitPrimarySources: Adding {} files from "
                                        "primary sources",
                                        *daq,
                                        result.result.size()));
            // @todo Confirm if we want to add parts even though state has transitioned past
            // Acquiring (which may mean manual stop and that parts have already been added)
            // @todo: How to treat errors from await?
            daq::AddDpParts(daq->GetContextMut(), result.result);
            daq->EmitContextSignal();
 
            // If daq is still acquiring we stop using strict error policy, otherwise do nothing
            if (!std::holds_alternative<Acquiring>(daq->m_state)) {
                LOG4CPLUS_DEBUG(
                    daq->m_logger,
                    fmt::format(
                        "{}: InitiateAwaitPrimarySources: "
                        "AwaitAsync completed but another operation has already transitioned "
                        "DAQ from Acquiring so automatic stop will not be performed.",
                        *daq));
                return;
            }
            daq->AddEvent<ActionEvent>(
                daq->GetId(),
                "DaqController::InitiateAwaitPrimarySources(): "
                "Primary sources completed. Performing automatic stop of metadata sources",
                daq->GetStatusRef().GetStatus());
 
            // Merge alerts from async op
            auto defer = ObservableStatus::DeferSignal(&daq->GetStatusRef());
            op::MergeAlerts(daq->GetStatusRef(), *alerts);
 
            // No continuation is necessary. If an error occurs that information is published
            // and user needs to intervene.
            // If a separate topic is created for errors only it may be published with a
            // continuation attached to the result of StopAsync (or more likely implemented inside
            // StopAsync).
            daq->StopAsync(ErrorPolicy::Strict);
        });
}
 
DpmDaqController::DpmDaqController(boost::asio::io_context& io_context,
                                   DaqContext context,
                                   std::shared_ptr<ObservableStatus> status,
                                   std::shared_ptr<ObservableEventLog> event_log,
                                   std::shared_ptr<DpmClient> dpm_client)
    : CommonDaqController(io_context, std::move(context), std::move(status), std::move(event_log))
    , m_liveness(std::make_shared<bool>(true))
    , m_dpm_client(std::move(dpm_client))
    , m_logger(log4cplus::Logger::getInstance("daq.ocm")) {
    assert(m_dpm_client);
    UpdateStateContext();
 
    // Connect slot to status signal which mirrors DAQ status of *this* from DPM
    //
    // Note: weak_from_this cannot be used in constructor as the reference count is not yet
    // incremented by the holding shared_ptr. For this reason the m_liveness is used instead.
    m_status_connection = m_dpm_client->ConnectStatusSignal(
        [id = GetId(), weak = std::weak_ptr<bool>(m_liveness), this](Status const& status) {
            if (id != status.id) {
                // Update for another DAQ
                return;
            }
            auto lock = weak.lock();
            if (!lock) {
                // Abandoned
                LOG4CPLUS_DEBUG("daq", "DpmDaqController is abandoned: " << this);
                return;
            }
            // Assign status of DAQ from DPM.
            LOG4CPLUS_DEBUG(m_logger, *this << ": Assigning new DAQ status from DPM: " << status);
            GetStatusRef() = status;
            UpdateStateContext();
        });
}
 
boost::future<State> DpmDaqController::StartAsync() {
    return boost::make_exceptional_future<State>(
        std::runtime_error(fmt::format("StartAsync() is invalid in state: {}", GetState())));
}
 
boost::future<Status> DpmDaqController::StopAsync(ErrorPolicy) {
    return boost::make_exceptional_future<Status>(
        std::runtime_error(fmt::format("StopAsync() is invalid in state: {}", GetState())));
}
 
boost::future<State>
DpmDaqController::AwaitAsync(std::vector<std::string>, std::chrono::milliseconds) {
    return boost::make_exceptional_future<State>(std::runtime_error(
        fmt::format("AwaitAsync() with sources is invalid in state: {}", GetState())));
}
 
void DpmDaqController::UpdateKeywords(fits::KeywordVector const&) {
    throw std::runtime_error(fmt::format("UpdateKeywords() is invalid in state: {}", GetState()));
}
 
State DpmDaqController::GetState() const DAQ_NOEXCEPT {
    return GetStatusRef().GetState();
}
 
boost::future<State> DpmDaqController::ScheduleMergeAsync() {
    LOG4CPLUS_TRACE(m_logger, *this << ": DpmDaqController::ScheduleMergeAsync");
 
    // If we are in NotScheduled and no request in-flight already: Send request to DPM to merge.
    // otherwise fail.
    auto starting_state = GetState();
    if (starting_state != State::NotScheduled) {
        return boost::make_exceptional_future<State>(std::runtime_error(
            fmt::format("ScheduleMergeAsync() is invalid in state: {}", GetState())));
    }
 
    assert(std::holds_alternative<NotScheduled>(m_state_ctx));
    auto& ctx = std::get<NotScheduled>(m_state_ctx);
    if (ctx.schedule_reply_pending) {
        // A request has already been sent
        return boost::make_exceptional_future<State>(
            std::logic_error("ScheduleMergeAsync() a request is already in flight"));
    }
    if (!m_dp_spec) {
        auto json = CreateDataProductSpecification(GetContext(), m_logger);
        m_dp_spec = json.dump(2);  // indent 2
    }
 
    return m_dpm_client->ScheduleAsync(*m_dp_spec)
        .then(GetIoExecutor(),
              [starting_state, weak = weak_from_this(), this](boost::future<State> f) -> State {
                  auto shared = weak.lock();
                  try {
                      auto state = f.get();
                      if (shared) {
                          // Update state only if DAQ has not changed state from starting_state.
                          //
                          // This can e.g. happen if OCM receive published state change data sample
                          // from DPM before receiving command reply.
                          if (GetState() == starting_state) {
                              SetState(state);
                          }
                          LOG4CPLUS_INFO(m_logger,
                                         fmt::format("{}: Scheduled DAQ successfully", *this));
                      } else {
                          LOG4CPLUS_DEBUG("daq", "DpmDaqController is abandoned: " << this);
                      }
                      return state;
                  } catch (elt::mal::TimeoutException const&) {
                      // Timeout errors does not result in error flag being set.
                      if (shared) {
                          LOG4CPLUS_INFO(m_logger,
                                         fmt::format("{}: Schedule DAQ failed with timeout "
                                                     "(no error flag set for this condition)",
                                                     *this));
                      }
                      throw;
                  } catch (...) {
                      if (shared) {
                          GetStatusRef().SetError(true);
                          LOG4CPLUS_ERROR(m_logger, fmt::format("{}: Scheduled DAQ failed", *this));
                      }
                      throw;
                  }
              });
}
 
boost::future<Status> DpmDaqController::AbortAsync(ErrorPolicy policy) {
    LOG4CPLUS_TRACE(m_logger, *this << ": DpmDaqController::AbortAsync");
 
    // 1. If we are sure DAQ has not been scheduled we can abort immediately.
    auto state = GetState();
    if (state == State::NotScheduled) {
        assert(std::holds_alternative<NotScheduled>(m_state_ctx));
        if (!std::get<NotScheduled>(m_state_ctx).schedule_reply_pending) {
            SetState(State::Aborted);
            return boost::make_ready_future<Status>(*GetStatus());
        }
    }
 
    // 2. Otherwise we must delegate to DPM to abort (this is also why we are not setting state
    //    AbortingMerging as we cannot know until we communiate with DPM).
    return m_dpm_client->AbortAsync(GetId()).then(
        GetIoExecutor(), [policy, weak = weak_from_this(), this](boost::future<State> f) -> Status {
            auto shared = weak.lock();
            if (!shared) {
                // Abandonded
                LOG4CPLUS_DEBUG("daq", "DpmDaqController is abandoned: " << this);
                throw boost::enable_current_exception(std::runtime_error("Operation aborted"));
            }
            try {
                auto result = f.get();
                SetState(result);
 
                return *GetStatus();
            } catch (...) {
                // Even if DPM failed we mark it as aborted if forced
                if (policy == ErrorPolicy::Tolerant) {
                    LOG4CPLUS_ERROR(m_logger,
                                    *this
                                        << ": Request to abort DAQ to DPM returned with "
                                           "error. ErrorPolicy::Tolerant is used so DAQ is marked "
                                           "as aborted anyway");
                    SetState(State::Aborted, true);
                    return *GetStatus();
                }
                throw;
            }
        });
}
 
void DpmDaqController::UpdateStateContext() {
    auto state = GetState();
    switch (state) {
    case State::NotScheduled:
        m_state_ctx = NotScheduled{false};
        return;
    default:
        m_state_ctx = std::monostate();
    };
}
 
void DpmDaqController::SetState(State state, std::optional<bool> error) {
    auto prev = GetState();
    GetStatusRef().SetState(state, error);
    if (prev != state) {
        UpdateStateContext();
    }
}
 
nlohmann::json CreateDataProductSpecification(DaqContext const& ctx, log4cplus::Logger& logger) {
    // @todo: At this time it is only a simplified version of the full capability
    // of the Data Product Specification without support for :
    // - in-place target
    // - keyword rules
    //
    // Current simplified source order:
    // 1. OCM keywords (ctx.keywords)
    // 2. Order of ctx.results
    nlohmann::json json;
    json["id"] = ctx.id;
    auto sources = nlohmann::json::array();
 
    if (!ctx.keywords.empty()) {
        // OCM keywords
        auto kws = fits::SerializeJsonKeywords(ctx.keywords);
        sources.push_back(nlohmann::json::object({{"type", "fitsKeywords"},
                                                  {"sourceName", ctx.process_name},
                                                  {"keywords", std::move(kws)}})
 
        );
    }
 
    // Temporary heuristics until OCM interface allows the explicit specification:
    // If
    // - number of primary sources == 1
    // - number of files from that primary  == 1
    // then we automatically designate it the *in-place* target.
    std::optional<nlohmann::json> target;
    std::optional<std::string> target_source_name;
 
    if (1 == ctx.prim_sources.size() &&
        1 == std::count_if(ctx.results.begin(),
                           ctx.results.end(),
                           [source = ctx.prim_sources[0].name](DpPart const& part) {
                               return part.origin == source;
                           })) {
        auto it = std::find_if(ctx.results.begin(),
                               ctx.results.end(),
                               [source = ctx.prim_sources[0].name](DpPart const& part) {
                                   return part.origin == source;
                               });
        assert(it != ctx.results.end());
        if (std::holds_alternative<std::string>(it->info)) {
            // At this point we have:
            // 1 primary source
            // 1 output fits file
            auto const& path = std::get<std::string>(it->info);
            target_source_name = ctx.prim_sources[0].name;
            LOG4CPLUS_DEBUG(logger,
                            fmt::format("{}: Heuristics resulted in using the file "
                                        "{} from {} as *in-place* merge target.",
                                        ctx.id,
                                        path,
                                        *target_source_name));
            target = nlohmann::json::object(
                {{"type", "fitsFile"}, {"sourceName", it->origin}, {"origin", path}});
        }
    }
 
    for (DpPart const& r : ctx.results) {
        if (target_source_name && *target_source_name == r.origin) {
            // Skip target file
            continue;
        }
        if (std::holds_alternative<fits::KeywordVector>(r.info)) {
            // keywords
            auto kws = fits::SerializeJsonKeywords(std::get<fits::KeywordVector>(r.info));
            sources.push_back(nlohmann::json::object(
                {{"type", "fitsKeywords"}, {"sourceName", r.origin}, {"keywords", std::move(kws)}})
 
            );
        } else if (std::holds_alternative<std::string>(r.info)) {
            auto const& path = std::get<std::string>(r.info);
            sources.push_back(nlohmann::json::object(
                {{"type", "fitsFile"}, {"sourceName", r.origin}, {"origin", path}})
 
            );
        }
    }
    json["sources"] = std::move(sources);
    json["target"] =
        nlohmann::json::object({{"fileId", ctx.file_id}, {"filePrefix", ctx.dp_name_prefix}});
 
    if (target) {
        json["target"]["source"] = *target;
    }
    return json;
}
 
}  // namespace daq