ocmDaqService.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_ocm_server
 * @copyright 2022 ESO - European Southern Observatory
 *
 * @brief Declaration of OcmDaqService
 */
#include "ocmDaqService.hpp"
 
#include <algorithm>
#include <cstddef>
#include <regex>
 
#include <fmt/format.h>
#include <fmt/ostream.h>
#include <log4cplus/loggingmacros.h>
#include <mal/rr/qos/ConnectionTime.hpp>
#include <nlohmann/json.hpp>
 
#include <daq/conversion.hpp>
#include <daq/daqController.hpp>
#include <daq/error/report.hpp>
#include <daq/fits/json.hpp>
#include <daqif/state.hpp>
 
#include <daq/json/startDaqV2.hpp>
 
#include "logger.hpp"
 
using boost::enable_current_exception;
using boost::make_exceptional_future;
using daq::error::NestedExceptionReporter;
using fmt::format;
using std::chrono::duration_cast;
using std::chrono::milliseconds;
 
namespace {
 
auto MakeDaqifException(std::string const& id,
                        std::string const& msg,
                        std::exception const& exception) {
    auto nested_msg = NestedExceptionReporter(exception).Str();
    if (msg.empty()) {
        return boost::enable_current_exception(
            daqif::DaqException(id, fmt::format("{}", nested_msg)));
 
    } else {
        return boost::enable_current_exception(
            daqif::DaqException(id, fmt::format("{}\n{}", msg, nested_msg)));
    }
}
 
std::vector<daq::DaqContext::Source> ParseSource(std::string const& str) {
    std::vector<daq::DaqContext::Source> sources;
 
    std::vector<ParsedSource> raw_sources = ParseSourceUris(str);
    sources.reserve(raw_sources.size());
    for (auto const& raw : raw_sources) {
        sources.push_back({raw.name, raw.rr_uri});
    }
 
    return sources;
}
 
void ValidateFilePrefix(char const* file_prefix) {
    auto file_regex = std::regex(R"(^[-a-zA-Z0-9_\.]*$)");
 
    if (!std::regex_match(file_prefix, file_regex)) {
        throw std::invalid_argument(
            fmt::format("file_prefix \"{}\" contains illegal characters, allowed: [a-zA-Z-0-9-_.]",
                        file_prefix));
    }
}
 
/**
 * Validate arguments.
 *
 * TODO:
 * - Check merge target.
 * - Source names must be unique (and shouldn't use reserved names)
 *
 * @throw std::invalid_argument on validation error.
 */
void ValidateDaqContext(daq::DaqContext const& ctx) {
    ValidateFilePrefix(ctx.dp_name_prefix.c_str());
}
 
}  // namespace
 
daq::DaqContext ParseStartDaqContext(std::string const& json_properties) {
    using std::chrono::duration_cast;
    using std::chrono::milliseconds;
    using Seconds = std::chrono::duration<double>;
    daq::DaqContext properties;
 
    if (json_properties.empty()) {
        // No arguments
        return properties;
    }
 
    auto json = nlohmann::json::parse(json_properties);
    if (!json.is_object()) {
        throw boost::enable_current_exception(std::invalid_argument(
            fmt::format("expected type object but got type {}", json.type_name())));
    }
    if (json.contains("keywords")) {
        properties.keywords = daq::fits::ParseJsonKeywords(json["keywords"]);
    }
    if (json.contains("awaitInterval")) {
        auto& value = json["awaitInterval"];
        if (!value.is_number()) {
            throw boost::enable_current_exception(std::invalid_argument(
                fmt::format("'awaitInterval': unsupported type: {}", value.type_name())));
        }
        auto await_interval = value.get<double>();
        if (await_interval < 0.0) {
            throw boost::enable_current_exception(std::invalid_argument(
                fmt::format("'awaitInterval' must be positive number, got {}", await_interval)));
        }
        properties.await_interval = duration_cast<milliseconds>(Seconds(value.get<double>()));
    }
    return properties;
}
 
bool ParsedSource::operator==(ParsedSource const& rhs) const {
    return name == rhs.name && rr_uri == rhs.rr_uri;
}
 
ParsedSource::ParsedSource(std::string name, std::string rr_uri)
    : name(std::move(name)), rr_uri(std::move(rr_uri)) {
}
std::ostream& operator<<(std::ostream& os, ParsedSource const& s) {
    os << "name: '" << s.name << "', rr_uri='" << s.rr_uri << "'";
    return os;
}
 
ParsedSource ParseSourceUri(std::string_view s) {
    auto start = s.find_first_not_of(' ');
    auto name_end_pos = s.find_first_of('@');
 
    if (name_end_pos == std::string_view::npos) {
        throw boost::enable_current_exception(
            std::invalid_argument("separator '@' not found in expected format 'name@rr-uri'"));
    }
    auto name = s.substr(start, name_end_pos - start);
    if (name.empty()) {
        throw boost::enable_current_exception(
            std::invalid_argument("name part in 'name@rr-uri' is empty"));
    }
 
    start = name_end_pos + 1;
    if (start >= s.size()) {
        throw boost::enable_current_exception(
            std::invalid_argument("invalid format string, expected format 'name@rr-uri'"));
    }
    auto rr_uri_end_pos = s.find_first_of(" ,", start);
    if (name_end_pos == std::string_view::npos) {
        throw boost::enable_current_exception(std::invalid_argument("separator ',' not found"));
    }
 
    auto rr_uri = s.substr(start, rr_uri_end_pos - start);
    if (rr_uri.empty()) {
        throw boost::enable_current_exception(
            std::invalid_argument("rr_uri part in 'name@rr-uri' is empty"));
    }
    return ParsedSource(std::string(name), std::string(rr_uri));
}
/**
 * Parse user provided string in the format
 * "<name>@<rr-uri>[ <name>@...]"
 *
 * @throw std::invalid_argument on errors.
 */
std::vector<ParsedSource> ParseSourceUris(std::string_view s) {
    std::vector<ParsedSource> result;
    std::size_t begin = 0;
 
    while (begin < s.size()) {
        const auto end = s.find_first_of(' ', begin);
 
        if (begin != end) {
            result.emplace_back(ParseSourceUri(s.substr(begin, end - begin)));
        }
 
        if (end == std::string_view::npos) {
            break;
        }
 
        begin = end + 1;
    }
 
    return result;
}
 
OcmDaqService::OcmDaqService(boost::asio::io_context& io_ctx,
                             mal::Mal& mal,
                             daq::Manager& mgr,
                             std::string proc_name,
                             std::string output_path,
                             std::shared_ptr<daq::ObservableEventLog> event_log)
    : m_io_ctx(io_ctx)
    , m_executor(m_io_ctx)
    , m_mal(mal)
    , m_mgr(mgr)
    , m_proc_name(std::move(proc_name))
    , m_output_path(std::move(output_path))
    , m_event_log(std::move(event_log))
    , m_log_observer_connection()
    , m_log_observer(log4cplus::Logger::getInstance(server::LOGGER_NAME_EVENTLOG))
    , m_logger(log4cplus::Logger::getInstance(server::LOGGER_NAME)) {
    m_log_observer_connection =
        m_event_log->ConnectObserver(std::reference_wrapper(m_log_observer));
    if (m_proc_name.empty()) {
        throw boost::enable_current_exception(
            std::invalid_argument("OcmDaqService: Process name cannot be empty"));
    }
}
 
OcmDaqService::~OcmDaqService() {
    m_log_observer_connection.disconnect();
}
 
boost::future<std::shared_ptr<::daqif::DaqReply>>
OcmDaqService::StartDaq(const std::string& id,
                        const std::string& file_prefix,
                        const std::string& primary_sources,
                        const std::string& metadata_sources,
                        const std::string& json_properties) {
    return boost::async(
               m_executor,
               [=, self = shared_from_this()]() mutable {
                   self->m_event_log->AddEvent(daq::UserActionEvent(
                       id,
                       fmt::format("Request received: "
                                   "StartDaq(id='{0}', file_prefix='{1}', "
                                   "primary_sources='{2}', metadata_sources='{3}', "
                                   "json_properties='{4}'",
                                   id,
                                   file_prefix,
                                   primary_sources,
                                   metadata_sources,
                                   json_properties),
                       std::nullopt));
                   try {
                       ValidateFilePrefix(file_prefix.c_str());
                   } catch (std::exception const& e) {
                       self->m_event_log->AddEvent(
                           daq::ErrorEvent(id, e.what(), std::nullopt, "user"));
                       return boost::make_exceptional_future<std::shared_ptr<::daqif::DaqReply>>(
                           daqif::DaqException(id, e.what()));
                   }
                   // Parse provided JSON
                   // Validation that require state is performed in m_executor for thread safety.
                   daq::DaqContext context;
                   try {
                       context = ParseStartDaqContext(json_properties);
                   } catch (std::exception const& e) {
                       auto msg =
                           fmt::format("Failed to parse StartDaq JSON properties: {}", e.what());
                       self->m_event_log->AddEvent(daq::ErrorEvent(id, msg, std::nullopt, "user"));
                       return boost::make_exceptional_future<std::shared_ptr<::daqif::DaqReply>>(
                           daqif::DaqException(id, msg));
                   }
 
                   context.file_id = self->m_mgr.MakeDaqId(&context.creation_time);
                   auto validated_id = id;
                   if (validated_id.empty()) {
                       // User did not provide an ID -> use file_id as DAQ id
                       validated_id = context.file_id;
                       LOG4CPLUS_INFO(
                           self->m_logger,
                           fmt::format("StartDaq(id='{0}'): Created and assigned DAQ id='{0}'",
                                       context.file_id));
                   } else {
                       // Check that an instance does not already exist with id
                       if (self->m_mgr.HaveDaq(validated_id)) {
                           LOG4CPLUS_INFO(
                               self->m_logger,
                               fmt::format("StartDaq(id='{0}'): DAQ with id='{0}' already exist",
                                           validated_id));
                           return boost::make_exceptional_future<std::shared_ptr<daqif::DaqReply>>(
                               daqif::DaqException(id,
                                                   "Data acquisition with same id already exist"));
                       }
                   }
 
                   try {
                       context.id = validated_id;
                       context.process_name = m_proc_name;
                       context.dp_name_prefix = file_prefix;
                       // Create primary sources
                       context.prim_sources = ParseSource(primary_sources);
                       // Create metadata sources
                       context.meta_sources = ParseSource(metadata_sources);
 
                       // Start
                       return self->StartDaq(context, "StartDaq");
                   } catch (daqif::DaqException const&) {
                       // Already correct exception
                       throw;
                   } catch (std::invalid_argument const& e) {
                       LOG4CPLUS_INFO(self->m_logger,
                                      fmt::format("StartDaq(id='{}'): Invalid argument error while "
                                                  "processing request: {}",
                                                  validated_id,
                                                  e.what()));
                       return boost::make_exceptional_future<std::shared_ptr<daqif::DaqReply>>(
                           daqif::DaqException(validated_id, e.what()));
                   } catch (std::exception const& e) {
                       LOG4CPLUS_INFO(self->m_logger,
                                      fmt::format("StartDaq(id='{}'): Error while"
                                                  "processing request: {}",
                                                  validated_id,
                                                  e.what()));
                       return boost::make_exceptional_future<std::shared_ptr<daqif::DaqReply>>(
                           daqif::DaqException(validated_id, e.what()));
                   } catch (...) {
                       LOG4CPLUS_INFO(
                           self->m_logger,
                           fmt::format("StartDaq(id='{}'): Unknown error while processing request",
                                       validated_id));
                       return boost::make_exceptional_future<std::shared_ptr<daqif::DaqReply>>(
                           daqif::DaqException(validated_id, "Uknown error"));
                   }
               })
        // unwrap outer future from async() to get the future we want from m_mgr.StartDaqAsync()
        .unwrap();
}
 
boost::future<std::shared_ptr<::daqif::DaqReply>>
OcmDaqService::StartDaqV2(const std::string& specification) {
    return boost::async(m_executor,
                        [=, self = shared_from_this()]() mutable {
                            self->m_event_log->AddEvent(daq::UserActionEvent(
                                "",
                                fmt::format("Request received: "
                                            "StartDaqV2(specification=\n'{0}')",
                                            specification),
                                std::nullopt));
                            try {
                                // Parse provided JSON
                                auto parsed = daq::json::ParseStartDaqV2Spec(
                                    nlohmann::json::parse(specification));
 
                                // Validation that require state is performed in m_executor for
                                // thread safety.
                                daq::DaqContext context;
                                UpdateFrom(context, parsed);
 
                                assert(context.specification.has_value());
 
                                ValidateDaqContext(context);
 
                                {
                                    nlohmann::json j;
                                    to_json(j, *context.specification);
                                    LOG4CPLUS_DEBUG(self->m_logger,
                                                    "Resulting specification after parsing: \n"
                                                        << j.dump(2));
                                }
 
                                // Start
                                return self->StartDaq(context, "StartDaqV2");
                            } catch (daqif::DaqException const&) {
                                // Already correct exception
                                throw;
                            } catch (nlohmann::json::exception const& e) {
                                throw MakeDaqifException("", "JSON parsing error", e);
                            } catch (daq::json::SchemaError const& e) {
                                throw MakeDaqifException("", "JSON Schema error", e);
                            } catch (std::invalid_argument const& e) {
                                throw MakeDaqifException("", "Invalid argument", e);
                            } catch (std::exception const& e) {
                                throw MakeDaqifException("", "", e);
                            }
                        })
        .unwrap();
}
 
boost::future<std::shared_ptr<::daqif::DaqReply>> OcmDaqService::StopDaq(const std::string& id) {
    return boost::async(
               m_executor,
               [self = shared_from_this(), id]() {
                   self->m_event_log->AddEvent(daq::UserActionEvent(id,
                                                                    fmt::format("Request received: "
                                                                                "StopDaq(id='{0}')",
                                                                                id),
                                                                    std::nullopt));
                   return self->StopDaq(id, false);
               })
        .unwrap();
}
 
boost::future<std::shared_ptr<::daqif::DaqReply>>
OcmDaqService::ForceStopDaq(const std::string& id) {
    return boost::async(m_executor,
                        [self = shared_from_this(), id]() {
                            self->m_event_log->AddEvent(
                                daq::UserActionEvent(id,
                                                     fmt::format("Request received: "
                                                                 "ForceStopDaq(id='{0}')",
                                                                 id),
                                                     std::nullopt));
                            return self->StopDaq(id, true);
                        })
        .unwrap();
}
 
boost::future<std::shared_ptr<::daqif::DaqReply>>
OcmDaqService::StartDaq(daq::DaqContext const& context, char const* function) {
    return m_mgr.StartDaqAsync(context).then(
        m_executor,
        [function, weak_self = std::weak_ptr(shared_from_this()), id = context.id](
            boost::future<daq::State> f) -> std::shared_ptr<daqif::DaqReply> {
            auto self = weak_self.lock();
            if (!self) {
                LOG4CPLUS_WARN(LOGGER_NAME,
                               fmt::format("{}(id='{}'): StartDaqAsync is "
                                           "complete but MAL service has "
                                           "been abandoned. Throwing exception.",
                                           function,
                                           id));
                throw boost::enable_current_exception(
                    daqif::DaqException(id, "Service has been abandoned"));
            }
            try {
                f.get();
                auto rep = self->m_mal.createDataEntity<daqif::DaqReply>();
                assert(rep);
                rep->setId(id);
                rep->setError(false);
                return rep;
            } catch (...) {
                auto what = self->MakeExceptionMessageWithStatus(id, std::current_exception());
                LOG4CPLUS_ERROR(self->m_logger,
                                fmt::format("{}(id='{}'): StartDaqAsync "
                                            "completed with failure: {}",
                                            function,
                                            id,
                                            what));
                throw boost::enable_current_exception(
                    daqif::DaqException(id, fmt::format("{}() failed: {}", function, what)));
            }
        });
}
 
boost::future<std::shared_ptr<::daqif::DaqReply>>
OcmDaqService::StopDaq(const std::string& id, bool forced) {
    return boost::async(
               m_executor,
               [self = shared_from_this(), id, forced]() {
                   return self->m_mgr
                       .StopDaqAsync(id,
                                     forced ? daq::ErrorPolicy::Tolerant : daq::ErrorPolicy::Strict)
                       .then(
                           self->m_executor,
                           [weak_self = std::weak_ptr(self->shared_from_this()),
                            id](boost::future<daq::Status> f) -> std::shared_ptr<daqif::DaqReply> {
                               auto self = weak_self.lock();
                               if (!self) {
                                   LOG4CPLUS_WARN(LOGGER_NAME,
                                                  fmt::format("StopDaq(id='{}'): StopDaqAsync is "
                                                              "complete but MAL service has "
                                                              "been abandoned. Throwing exception.",
                                                              id));
                                   throw boost::enable_current_exception(
                                       daqif::DaqException(id, "Service has been abandoned"));
                               }
                               try {
                                   f.get();
                                   auto rep = self->m_mal.createDataEntity<daqif::DaqReply>();
                                   assert(rep);
                                   rep->setId(id);
                                   rep->setError(false);
                                   return rep;
                               } catch (...) {
                                   auto what = self->MakeExceptionMessageWithStatus(
                                       id, std::current_exception());
                                   LOG4CPLUS_INFO(self->m_logger,
                                                  fmt::format("StopDaq(id='{}'): "
                                                              "completed with failure: {}",
                                                              id,
                                                              what));
                                   throw boost::enable_current_exception(daqif::DaqException(
                                       id, fmt::format("Stop failed\n\n{}", what)));
                               }
                           });
               })
        // unwrap outer future from async() to get the future we want from m_mgr.StopDaqAsync()
        .unwrap();
}
 
boost::future<std::shared_ptr<::daqif::DaqReply>> OcmDaqService::AbortDaq(const std::string& id) {
    return boost::async(m_executor,
                        [self = shared_from_this(), id]() {
                            self->m_event_log->AddEvent(
                                daq::UserActionEvent(id,
                                                     fmt::format("Request received: "
                                                                 "AbortDaq(id='{0}')",
                                                                 id),
                                                     std::nullopt));
                            return self->AbortDaq(id, false);
                        })
        .unwrap();
}
 
boost::future<std::shared_ptr<::daqif::DaqReply>>
OcmDaqService::ForceAbortDaq(const std::string& id) {
    return boost::async(m_executor,
                        [self = shared_from_this(), id]() {
                            self->m_event_log->AddEvent(
                                daq::UserActionEvent(id,
                                                     fmt::format("Request received: "
                                                                 "ForceAbortDaq(id='{0}')",
                                                                 id),
                                                     std::nullopt));
                            return self->AbortDaq(id, true);
                        })
        .unwrap();
}
 
boost::future<std::shared_ptr<::daqif::DaqReply>>
OcmDaqService::AbortDaq(const std::string& id, bool forced) {
    return boost::async(
               m_executor,
               [self = shared_from_this(), id, forced]() {
                   return self->m_mgr
                       .AbortDaqAsync(
                           id, forced ? daq::ErrorPolicy::Tolerant : daq::ErrorPolicy::Strict)
                       .then(
                           self->m_executor,
                           [weak_self = std::weak_ptr(self->shared_from_this()), id, forced](
                               boost::future<daq::Status> f) -> std::shared_ptr<daqif::DaqReply> {
                               auto self = weak_self.lock();
                               if (!self) {
                                   LOG4CPLUS_WARN(
                                       LOGGER_NAME,
                                       fmt::format("AbortDaq(id='{}', forced={}): AbortDaqAsync is "
                                                   "complete but MAL service has "
                                                   "been abandoned. Throwing exception.",
                                                   id,
                                                   forced));
                                   throw boost::enable_current_exception(
                                       daqif::DaqException(id, "Service has been abandoned"));
                               }
                               try {
                                   auto result = f.get();
                                   LOG4CPLUS_INFO(
                                       self->m_logger,
                                       fmt::format("AbortDaq(id='{}', forced={}): "
                                                   "AbortDaqAsync Completed successfully",
                                                   id,
                                                   forced));
                                   auto rep = self->m_mal.createDataEntity<daqif::DaqReply>();
                                   assert(rep);
                                   rep->setId(id);
                                   rep->setError(HasError(result));
                                   LOG4CPLUS_DEBUG(
                                       self->m_logger,
                                       fmt::format("AbortDaq(id='{}', forced={}): "
                                                   "AbortDaqAsync Completed, returning reply now.",
                                                   id,
                                                   forced));
                                   return rep;
                               } catch (...) {
                                   auto what = self->MakeExceptionMessageWithStatus(
                                       id, std::current_exception());
                                   LOG4CPLUS_ERROR(self->m_logger,
                                                   fmt::format("AbortDaq(id='{}', forced={}): "
                                                               "AbortDaqAsync Completed "
                                                               "with fatal error:\n{}",
                                                               id,
                                                               forced,
                                                               what));
                                   throw boost::enable_current_exception(daqif::DaqException(
                                       id, fmt::format("Abort failed\n\n{}", what)));
                               }
                           });  // m_msg.AbortDaqAsync cont
               })               // boost::async()
        // unwrap outer future from async() to get the future we want from m_mgr.AbortDaqAsync()
        .unwrap();
}
 
boost::future<std::shared_ptr<::daqif::DaqReply>>
OcmDaqService::UpdateKeywords(const std::string& id, const std::string& keywords) {
    return boost::async(
        m_executor,
        [self = shared_from_this(), id, keywords]() -> std::shared_ptr<::daqif::DaqReply> {
            self->m_event_log->AddEvent(
                daq::UserActionEvent(id,
                                     fmt::format("Request received: "
                                                 "UpdateKeywords(id='{0}', keywords='{1}')",
                                                 id,
                                                 keywords),
                                     std::nullopt));
            daq::fits::KeywordVector parsed_keywords;
            try {
                daq::fits::ParseJsonKeywords(keywords.c_str()).swap(parsed_keywords);
            } catch (nlohmann::json::exception const& e) {
                LOG4CPLUS_ERROR(
                    self->m_logger,
                    fmt::format("UpdateKeywords(id='{}', ...): Failed to parse JSON", id));
                throw boost::enable_current_exception(
                    daqif::DaqException(id, fmt::format("Invalid JSON string: {}", e.what())));
            } catch (std::invalid_argument const& e) {
                LOG4CPLUS_ERROR(
                    self->m_logger,
                    fmt::format(
                        "UpdateKeywords(id='{}', ...): JSON could be parsed but was invalid "
                        "schema",
                        id));
                throw boost::enable_current_exception(
                    daqif::DaqException(id, fmt::format("Invalid JSON schema: {}", e.what())));
            } catch (std::exception const& e) {
                LOG4CPLUS_ERROR(
                    self->m_logger,
                    fmt::format(
                        "UpdateKeywords(id='{}', ...): std::exception: '{}'", id, e.what()));
                throw boost::enable_current_exception(
                    daqif::DaqException(id, fmt::format("std::exception: {}", e.what())));
            } catch (...) {
                throw boost::enable_current_exception(daqif::DaqException(id, "unknown error"));
            }
            try {
                self->m_mgr.UpdateKeywords(id, parsed_keywords);
                auto rep = self->m_mal.createDataEntity<daqif::DaqReply>();
                rep->setId(id);
                rep->setError(false);
                return rep;
            } catch (daq::fits::InvalidKeyword const& e) {
                LOG4CPLUS_ERROR(self->m_logger,
                                fmt::format("UpdateKeywords(id='{}'): {}", id, e.what()));
                throw boost::enable_current_exception(daqif::DaqException(id, e.what()));
            } catch (daq::InvalidDaqId const& e) {
                LOG4CPLUS_ERROR(
                    self->m_logger,
                    fmt::format(
                        "UpdateKeywords(id='{}'): Invalid data acquisition id: ", id, e.what()));
                throw boost::enable_current_exception(daqif::DaqException(id, e.what()));
            } catch (std::exception const& e) {
                LOG4CPLUS_ERROR(
                    self->m_logger,
                    fmt::format(
                        "UpdateKeywords(id='{}', ...): std::exception: '{}'", id, e.what()));
                throw boost::enable_current_exception(
                    daqif::DaqException(id, fmt::format("std::exception: {}", e.what())));
            } catch (...) {
                throw boost::enable_current_exception(daqif::DaqException(id, "unknown error"));
            }
        });
}
 
boost::future<std::shared_ptr<::daqif::DaqStatus>> OcmDaqService::GetStatus(const std::string& id) {
    return boost::async(
               m_executor,
               [self = shared_from_this(), id]() {
                   self->m_event_log->AddEvent(
                       daq::UserActionEvent(id,
                                            fmt::format("Request received: "
                                                        "GetStatus(id='{0}')",
                                                        id),
                                            std::nullopt));
                   try {
                       LOG4CPLUS_INFO(self->m_logger, fmt::format("GetStatus(id='{}'): Enter", id));
                       auto status = self->m_mgr.GetStatus(id);
                       auto rep = self->m_mal.createDataEntity<daqif::DaqStatus>();
                       assert(rep);
                       *rep << status;
                       LOG4CPLUS_INFO(
                           self->m_logger,
                           fmt::format("GetStatus(id='{}'): Set result -> {}", id, status.state));
                       return boost::make_ready_future<std::shared_ptr<daqif::DaqStatus>>(rep);
                   } catch (std::invalid_argument const&) {
                       LOG4CPLUS_ERROR(
                           self->m_logger,
                           fmt::format("GetStatus(id='{}'): Invalid data acquisition id", id));
                       return boost::make_exceptional_future<std::shared_ptr<daqif::DaqStatus>>(
                           boost::enable_current_exception(daqif::DaqException(
                               id, fmt::format("No data acquisition with id='{}'", id))));
                   }
               })
        .unwrap();
}
 
boost::future<std::vector<std::shared_ptr<::daqif::DaqStatus>>> OcmDaqService::GetActiveList() {
    return boost::async(
        m_executor,
        [self = shared_from_this()]() -> std::vector<std::shared_ptr<::daqif::DaqStatus>> {
            self->m_event_log->AddEvent(daq::UserActionEvent("",
                                                             "Request received: "
                                                             "GetActiveList()",
                                                             std::nullopt));
            auto daqs = self->m_mgr.GetDaqControllers();
            std::vector<std::shared_ptr<daq::DaqController const>> active;
            std::vector<std::shared_ptr<daqif::DaqStatus>> reply;
            std::copy_if(daqs.begin(), daqs.end(), std::back_inserter(active), [](auto daq_ctl) {
                return !daq::IsFinalState(daq_ctl->GetState());
            });
            std::transform(active.begin(),
                           active.end(),
                           std::back_inserter(reply),
                           [&mal = self->m_mal](auto daq_ctl) {
                               auto mal_status = mal.createDataEntity<daqif::DaqStatus>();
                               *mal_status << daq_ctl->GetStatus()->GetStatus();
                               return mal_status;
                           });
            return reply;
        });
}
 
boost::future<std::shared_ptr<::daqif::AwaitDaqReply>> OcmDaqService::AwaitDaqState(
    const std::string& id, daqif::DaqState state, daqif::DaqSubState substate, double timeout) {
    using Seconds = std::chrono::duration<double>;
 
    return boost::async(
               m_executor,
               [=, self = shared_from_this()]() {
                   self->m_event_log->AddEvent(daq::UserActionEvent(id,
                                                                    format("Request received: "
                                                                           "AwaitDaqState(id='{}', "
                                                                           "state={}, substate={}, "
                                                                           "timeout={})",
                                                                           id,
                                                                           daq::ToString(state),
                                                                           daq::ToString(substate),
                                                                           timeout),
                                                                    std::nullopt));
                   if (timeout <= 0) {
                       return boost::make_exceptional_future<daq::Result<daq::Status>>(
                           std::invalid_argument(
                               format("Invalid argument `timeout`. Must be > 0", timeout)));
                   }
                   if (!daqif::IsStateValid(state, substate)) {
                       return boost::make_exceptional_future<daq::Result<daq::Status>>(
                           std::invalid_argument(fmt::format(
                               "Invalid state combination: {} and {}", state, substate)));
                   }
                   auto daq_state = daq::MakeState({state, substate});
                   return self->m_mgr.AwaitDaqStateAsync(
                       id, daq_state, duration_cast<milliseconds>(Seconds(timeout)));
               })
        .unwrap()
        .then(
            m_executor,
            [id, self = shared_from_this()](boost::future<daq::Result<daq::Status>> fut) {
                try {
                    auto [timeout, status] = fut.get();
                    auto mal_reply = self->m_mal.createDataEntity<daqif::AwaitDaqReply>();
                    assert(mal_reply);
                    mal_reply->setTimeout(timeout);
                    auto mal_status_ptr = mal_reply->getStatus();
                    assert(mal_status_ptr);
                    *mal_status_ptr << status;
                    daq::ActionEvent(id,
                                     fmt::format("Request completed: {}",
                                                 (timeout ? "condition not yet satisfied (timeout)"
                                                          : "condition satisfied")),
                                     std::nullopt);
                    return mal_reply;
                } catch (std::exception const& e) {
                    auto what = self->MakeExceptionMessageWithStatus(id, std::current_exception());
                    self->m_event_log->AddEvent(daq::ActionEvent(
                        id,
                        fmt::format("Await state completed exceptionally\n\n{}", what),
                        std::nullopt));
                    throw boost::enable_current_exception(
                        daqif::DaqException(id, fmt::format("Await state failed\n\n{}", what)));
                } catch (...) {
                    daq::ActionEvent(
                        id, "Request completed exceptionally: Unknown exception", std::nullopt);
                    throw boost::enable_current_exception(
                        daqif::DaqException(id, "Uknown exception"));
                }
            });
}
 
std::string
OcmDaqService::MakeExceptionMessageWithStatus(std::string const& id,
                                              std::exception_ptr const& exception) const {
    auto nested_msg = NestedExceptionReporter(exception).Str();
    auto alerts_msg = std::string("n/a");
    try {
        auto status = m_mgr.GetStatus(id);
        alerts_msg = fmt::format("{}", status.alerts);
    } catch (std::exception const& e) {
        LOG4CPLUS_WARN(m_logger, fmt::format("GetStatus({}) failed:", id, e.what()));
    }
    return fmt::format("Errors(s):\n{}\nActive alert(s):\n{}", nested_msg, alerts_msg);
}
 
void OcmDaqService::UpdateFrom(daq::DaqContext& context, daq::json::StartDaqV2Spec const& spec) {
    using PrimaryDataSource = daq::json::StartDaqV2Spec::PrimaryDataSource;
    using MetadataSource = daq::json::StartDaqV2Spec::MetadataSource;
    using FitsKeywordsSource = daq::json::FitsKeywordsSource;
    using FitsFileSource = daq::json::FitsFileSource;
 
    context.file_id = m_mgr.MakeDaqId(&context.creation_time);
    if (context.id.empty()) {
        context.id = !spec.id.empty() ? spec.id : context.file_id;
    }
 
    if (spec.await_completion_interval.has_value()) {
        context.await_interval = *spec.await_completion_interval;
    }
 
    context.process_name = m_proc_name;
    context.dp_name_prefix = spec.file_prefix;
    context.specification = spec;
 
    // Create sources
    for (auto const& variant : spec.sources) {
        if (auto const* ds = std::get_if<PrimaryDataSource>(&variant); ds != nullptr) {
            context.prim_sources.push_back({ds->source_name, ds->rr_uri});
        } else if (auto const* ds = std::get_if<MetadataSource>(&variant); ds != nullptr) {
            context.meta_sources.push_back({ds->source_name, ds->rr_uri});
        } else if (auto const* ds = std::get_if<FitsKeywordsSource>(&variant); ds != nullptr) {
            context.results.emplace_back(ds->source_name, ds->keywords);
        } else if (auto const* ds = std::get_if<FitsFileSource>(&variant); ds != nullptr) {
            context.results.emplace_back(ds->source_name, ds->location);
        } else {
            LOG4CPLUS_ERROR(m_logger, "Unknown variant encountered");
        }
    }
 
    // Check that an instance does not already exist with id
    if (m_mgr.HaveDaq(context.id)) {
        auto msg = fmt::format("DAQ with id='{0}' already exist", context.id);
        LOG4CPLUS_INFO(m_logger, msg);
        throw boost::enable_current_exception(std::invalid_argument(msg));
    }
}