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/json.hpp>
#include <daq/makeDpSpec.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(15))},
                {}));
    }
 
    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(15))},
                {}));
    }
 
    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(rad::IoExecutor& ex)
    : executor(ex)
    , start([this](auto par) {
        return daq::op::InitiateOperation<daq::op::StartAsync>(executor, par);
    })
    , abort([this](ErrorPolicy policy, auto par) {
        return daq::op::InitiateOperation<op::AbortAsync>(executor, policy, std::move(par));
    })
    , stop([this](ErrorPolicy policy, auto par) {
        return daq::op::InitiateOperation<op::StopAsync>(executor, policy, std::move(par));
    })
    , await_prim([this](auto par) {
        return daq::op::InitiateAbortableOperation<op::AwaitPrimAsync>(executor, 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_executor(io_ctx)
    , m_mal(mal)
    , m_async_ops(m_executor)
    , 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,
                                               fits::KeywordFormatter const& kw_formatter)
    -> std::shared_ptr<DaqController> {
    DaqSources sources = MakeSources(m_mal, daq_ctx);
    return OcmDaqController::Create(m_io_ctx,
                                    kw_formatter,
                                    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->HasError();
}
 
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,
                         fits::KeywordFormatter const& kw_formatter,
                         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,
                                                                  kw_formatter,
                                                                  std::move(context),
                                                                  sources,
                                                                  std::move(status),
                                                                  std::move(event_log),
                                                                  std::move(ops)));
}
 
OcmDaqController::OcmDaqController(boost::asio::io_context& io_context,
                                   fits::KeywordFormatter const& kw_formatter,
                                   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_kw_formatter(kw_formatter)
    , 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.ocm.controller")) {
    if (GetContext().id != GetStatusRef().GetId()) {
        throw boost::enable_current_exception(
            std::invalid_argument("Data acquisition id mismatch between DaqContext "
                                  "and ObservableStatus"));
    }
 
    if (!HasStatefulSources() && GetContext().results.empty()) {
        throw boost::enable_current_exception(
            std::invalid_argument("No data sources or data to merge is 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::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,
                             m_kw_formatter);
}
 
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,
                                               m_kw_formatter),
                             GetContext().await_interval);
}
 
void OcmDaqController::AddInitialKeywords() {
    // Nothing yet.
}
 
auto OcmDaqController::HasStatefulSources() noexcept -> bool {
    return !(m_prim_sources.empty() && m_meta_sources.empty());
}
 
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()),
               this](boost::future<void> f) {
                  LOG4CPLUS_INFO(daq->GetLogger(),
                                 fmt::format("{}: StartAsync: Completed {}",
                                             *daq,
                                             f.has_value() ? "successfully" : "with error"));
                  auto& status = GetStatusRef();
                  auto alert = ObservableStatus::AlertActivator(
                      &status, alert::DAQ_CONTROLLER, "StartAsync()");
                  try {
                      // Merge alerts from async op
                      op::MergeAlerts(status, *alerts);
 
                      // Re-raise any exception
                      if (f.has_exception()) {
                          (void)f.get();
                      }
                      // Await completion if there are primary data sources
                      InitiateStopCondition();
                      // No exception -> all done! update and return Acquiring state
                      SetState(Acquiring{});
                      alert.Clear();
                      return GetState();
                  } catch (std::exception const& e) {
                      LOG4CPLUS_ERROR(m_logger,
                                      fmt::format("{}: AsyncStartDaq failed: {}", *this, e.what()));
                      alert.Set(fmt::format("Failed to start DAQ: {}", e.what()));
                      throw;
                  }
              });
}
 
boost::future<Status> OcmDaqController::StopAsync(ErrorPolicy policy) {
    GetEventLogRef().AddEvent(
        ActionEvent(GetId(), "DaqController::StopAsync()", GetStatusRef().GetStatus()));
 
    if (!(std::holds_alternative<Acquiring>(m_state) ||
          std::holds_alternative<Stopping>(m_state))) {
        return boost::make_exceptional_future<Status>(
            std::runtime_error(fmt::format("Data acquisition can only be stopped while Acquiring "
                                           "or being Stopped, current state: {}",
                                           GetState())));
    }
 
    // @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{});
 
    // As we're asked to stop manually we stop waiting for primary sources
    CancelAwaitPrimarySources();
 
    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()),
               this](boost::future<Result<DpParts>> f) mutable -> Status {
                  if (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(
                          GetLogger(),
                          fmt::format("{}: StopAsync: Data acquisition modified by other commands. "
                                      "Do nothing else (errors are ignored). ",
                                      *this));
                      throw std::runtime_error(fmt::format(
                          "Stop command could not be completed because Data Acquisitions state was "
                          "modified in the meantime (current state: {})",
                          GetState()));
                  }
                  LOG4CPLUS_INFO(GetLogger(),
                                 fmt::format("{}: StopAsync: Completed {}",
                                             *this,
                                             f.has_value() ? "successfully" : "with error"));
                  auto& status = GetStatusRef();
                  auto alert = ObservableStatus::AlertActivator(
                      &status, alert::DAQ_CONTROLLER, "StopAsync()");
                  try {
                      op::MergeAlerts(status, *alerts);
 
                      // Also may throw to propagate error
                      auto result = f.get();
                      if (result.error) {
                          alert.Set("Async operation Stop completed with error(s)");
                      } else {
                          alert.Clear();
                      }
                      AddDpParts(GetContextMut(), result.result);
                      EmitContextSignal();
 
                      // If there were no exceptions we're all done.
                      // Update and return Stopped state
                      SetState(Stopped{});
                      return GetStatus()->GetStatus();
                  } catch (std::exception const& e) {
                      LOG4CPLUS_ERROR(m_logger,
                                      fmt::format("{}: StopAsync failed: {}", *this, e.what()));
                      alert.Set(fmt::format("Failed to stop DAQ: {}", e.what()));
                      throw;
                  }
              });
}
 
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{});
 
    // As we're asked to stop manually we stop waiting for primary sources
    CancelAwaitPrimarySources();
 
    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()),
               this](boost::future<Result<void>> f) -> Status {
                  if (GetState() == State::Aborted) {
                      // It can happen that a request to abort was superseded and
                      // finalized before reply was received.
                      LOG4CPLUS_INFO(
                          GetLogger(),
                          fmt::format("{}: AbortAsync: Data acquisition already aborted. "
                                      "Do nothing else (errors are ignored). ",
                                      *this));
                      // @todo: Should throw instead as the command was superseeded by
                      // other command.
                      return GetStatus()->GetStatus();
                  }
                  LOG4CPLUS_DEBUG(GetLogger(),
                                  fmt::format("{}: AbortAsync: Completed, updating DAQ status and "
                                              "set reply remaining. Has fatal error={}",
                                              *daq,
                                              f.has_exception()));
                  auto& status = GetStatusRef();
                  // Merge alerts from saync op
                  auto alert = ObservableStatus::AlertActivator(
                      &status, alert::DAQ_CONTROLLER, "AbortAsync()");
                  try {
                      op::MergeAlerts(status, *alerts);
 
                      auto result = f.get();
                      if (result.error) {
                          alert.Set("Async operation Abort completed with error(s)");
                      } else {
                          alert.Clear();
                      }
                      // Success
                      daq->SetState(Aborted{});
                      LOG4CPLUS_INFO(GetLogger(),
                                     fmt::format("{}: AbortAsync: Completed successfully.", *daq));
                      return GetStatus()->GetStatus();
                  } catch (std::exception const& e) {
                      LOG4CPLUS_ERROR(m_logger,
                                      fmt::format("{}: AbortAsync failed: {}", *this, e.what()));
                      alert.Set(fmt::format("Failed to abort DAQ: {}", e.what()));
                      throw;
                  }
              });
}
 
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, m_kw_formatter);
    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<Source<PrimSource>*, 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)>::type::StateSignal::slot_type;
                s->ConnectStateListener(SlotType(std::move(listener)).track_foreign(daq));
            },
            source);
    }
    return promise->get_future();
}
 
std::optional<std::variant<Source<PrimSource>*, 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 &(*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 &(*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::InitiateStopCondition() {
    AddEvent<ActionEvent>(
        GetId(), "DaqController::InitiateStopCondition(): Initiating", GetStatusRef().GetStatus());
    if (!HasStatefulSources()) {
        LOG4CPLUS_DEBUG(
            m_logger,
            fmt::format("{}: InitiateStopCondition: No stateful sources to wait for -> stopping",
                        *this));
        // If we don't have any stateful data sources at all we stop automatically. This needs to be
        // done asynchronously though to match expectation that this is an async initiation
        // function.
        boost::async(GetIoExecutor(),
                     [daq = std::static_pointer_cast<OcmDaqController>(shared_from_this())] {
                         daq->StopAsync(ErrorPolicy::Strict);
                     });
        return;
    }
    if (m_prim_sources.empty()) {
        LOG4CPLUS_DEBUG(
            m_logger,
            fmt::format("{}: InitiateStopCondition: No primary sources to monitor.", *this));
        return;
    }
    LOG4CPLUS_DEBUG(m_logger, fmt::format("{}: InitiateStopCondition: 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) {
            // Remove abort callback as async op has completed
            daq->m_abort_await_primary_sources = nullptr;
 
            // Add FITS files from the stopped sources
            auto result = fut.get();
            LOG4CPLUS_DEBUG(daq->m_logger,
                            fmt::format("{}: InitiateStopCondition: 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(
                        "{}: InitiateStopCondition: "
                        "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::InitiateStopCondition(): "
                "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);
        });
}
 
void OcmDaqController::CancelAwaitPrimarySources() {
    if (m_abort_await_primary_sources) {
        LOG4CPLUS_TRACE(m_logger, "OcmDaqController::CancelAwaitPrimarySources()");
        m_abort_await_primary_sources();
        m_abort_await_primary_sources = nullptr;
    }
}
 
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.controller")) {
    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.
    //
    // Additionally the status provided may from a past update, so we only update if received status
    // is newer than current.
    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;
            }
 
            auto const& current = GetStatusRef().GetStatus();
            if (current.timestamp > status.timestamp) {
                LOG4CPLUS_TRACE(
                    m_logger,
                    "Ignoring DAQ status update: From DPM: " << status << ", current: " << current);
                return;
            }
            // Assign status of DAQ from DPM.
            LOG4CPLUS_TRACE(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 dp_spec = MakeDataProductSpecification(GetContext(), m_logger);
        nlohmann::json j;
        to_json(j, dp_spec);
        m_dp_spec = j.dump(2);  // indent 2
        LOG4CPLUS_DEBUG(m_logger, "Created DpSpec:\n" << *m_dp_spec);
    }
    auto status = std::string();
    {
        nlohmann::json j;
        j = GetStatus()->GetStatus();
        status = j.dump(2);
    }
    return m_dpm_client->ScheduleAsync(*m_dp_spec, status)
        .then(GetIoExecutor(),
              [starting_state, weak = weak_from_this(), this](boost::future<State> f) -> State {
                  auto alert_id = MakeAlertId(alert::REQUEST, "dpm.QueueDaq()");
                  auto shared = weak.lock();
                  try {
                      auto state = f.get();
                      if (shared) {
                          auto& status = GetStatusRef();
                          ObservableStatus::DeferSignal d(&status);
                          // 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.
                          // @todo: Use timestamp instead?
                          if (GetState() == starting_state) {
                              SetState(state);
                          }
                          status.ClearAlert(alert_id);
                          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 (std::exception const& e) {
                      if (shared) {
                          auto& status = GetStatusRef();
                          ObservableStatus::DeferSignal d(&status);
                          status.SetAlert(MakeAlert(
                              alert_id,
                              fmt::format("Failed to schedule DAQ for merging: {}", e.what())));
                          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.
    // @todo: Handle case where request to schedule has been sent, but no acknowledgement received.
    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).
    //    If abort fails but is forced to complete an alert is added.
    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"));
            }
            auto& status = GetStatusRef();
            auto alert =
                ObservableStatus::AlertActivator(&status, alert::REQUEST, "dpm.AbortAsync()");
            try {
                auto result = f.get();
                SetState(result);
                alert.Clear();
 
                return *GetStatus();
            } catch (std::exception const& e) {
                // Even if DPM failed we mark it as aborted if forced
                // Unless we force completion we don't add the alert to DAQ as DPM owns the state.
                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");
                    alert.Set(
                        fmt::format("Async operation Abort completed with error(s): {}", e.what()));
                    SetState(State::Aborted);
                    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) {
    auto prev = GetState();
    GetStatusRef().SetState(state);
    if (prev != state) {
        UpdateStateContext();
    }
}
 
}  // namespace daq