status.cpp

Go to the documentation of this file.

/**
 * @file
 * @ingroup daq_common_libdaq
 * @copyright
 * (c) Copyright ESO 2023
 * All Rights Reserved
 * ESO (eso.org) is an Intergovernmental Organisation, and therefore special legal conditions apply.
 */
#include <algorithm>
#include <ctime>
#include <daq/status.hpp>
#include <iomanip>
#include <ostream>
 
#include <boost/assert.hpp>
#include <fmt/format.h>
 
namespace daq {
 
void SetAlert(std::vector<Alert>& alerts, Alert alert) {
    auto it = find(alerts.begin(), alerts.end(), alert);
    if (it != alerts.end()) {
        // Replace
        *it = std::move(alert);
    } else {
        // Or add
        alerts.emplace_back(std::move(alert));
    }
}
 
bool ClearAlert(std::vector<Alert>& alerts, AlertId const& alert) {
    auto it =
        alert.key == "*"
            ? std::remove_if(alerts.begin(),
                             alerts.end(),
                             [&](Alert const& cmp) { return cmp.id.category == alert.category; })
            : std::remove(alerts.begin(), alerts.end(), alert);
    if (it != alerts.end()) {
        alerts.erase(it, alerts.end());
        return true;
    }
    return false;
}
 
Alert MakeAlert(std::string_view category, std::string key, std::string description) {
    return Alert{{std::string(category), std::move(key)},
                 std::move(description),
                 Alert::TimePoint::clock::now()};
}
 
Alert MakeAlert(AlertId id, std::string description) {
    return Alert{std::move(id), std::move(description), Alert::TimePoint::clock::now()};
}
 
AlertId MakeAlertId(std::string_view category, std::string key) {
    return AlertId{std::string(category), std::move(key)};
}
 
bool operator==(AlertId const& lhs, AlertId const& rhs) noexcept {
    return lhs.key == rhs.key && lhs.category == rhs.category;
}
 
bool operator!=(AlertId const& lhs, AlertId const& rhs) noexcept {
    return !(lhs == rhs);
}
 
std::ostream& operator<<(std::ostream& os, AlertId const& s) {
    os << s.category << "-" << s.key;
    return os;
}
 
bool operator==(Alert const& lhs, Alert const& rhs) noexcept {
    return lhs.id == rhs.id;
}
 
bool operator==(Alert const& lhs, AlertId const& rhs) noexcept {
    return lhs.id == rhs;
}
 
bool operator==(AlertId const& lhs, Alert const& rhs) noexcept {
    return lhs == rhs.id;
}
 
bool operator!=(Alert const& lhs, Alert const& rhs) noexcept {
    return !(lhs == rhs);
}
 
std::ostream& operator<<(std::ostream& os, PutTimeStruct const& t) {
    auto timet = std::chrono::system_clock::to_time_t(t.time);
    using milli = std::chrono::milliseconds;
 
    std::ios restore(nullptr);
    restore.copyfmt(os);
    os << std::put_time(std::gmtime(&timet), "%FT%T.") << std::setfill('0') << std::setw(3)
       << std::chrono::duration_cast<milli>(t.time.time_since_epoch()).count() % 1000;
 
    os.copyfmt(restore);
    return os;
}
 
PutTimeStruct PutTime(std::chrono::system_clock::time_point const& time) {
    return PutTimeStruct{time};
}
 
std::ostream& operator<<(std::ostream& os, Alert const& alert) {
    os << PutTime(alert.timestamp) << " (" << alert.id << "): " << alert.description;
    return os;
}
 
std::ostream& operator<<(std::ostream& os, std::vector<Alert> const& alerts) {
    auto count = 0u;
    for (auto const& alert : alerts) {
        os << "[" << count << "]: " << alert << "\n";
        ++count;
    }
    return os;
}
 
bool ReceiverStatus::IsFinalState() const noexcept {
    return state == ReceiverStatus::State::Success || state == ReceiverStatus::State::Failure;
}
 
bool operator==(ReceiverStatus const& lhs, ReceiverStatus const& rhs) noexcept {
    return lhs.state == rhs.state;
}
 
bool operator!=(ReceiverStatus const& lhs, ReceiverStatus const& rhs) noexcept {
    return !(lhs == rhs);
}
 
std::ostream& operator<<(std::ostream& os, ReceiverStatus::State state) {
    switch (state) {
    case ReceiverStatus::State::NotStarted:
        os << "NotStarted";
        break;
    case ReceiverStatus::State::Started:
        os << "Started";
        break;
    case ReceiverStatus::State::Success:
        os << "Success";
        break;
    case ReceiverStatus::State::Failure:
        os << "Failure";
        break;
    default:
        os << "Unknown";
        break;
    };
    return os;
}
 
Status::Status(std::string id, std::string file_id, State state, TimePoint timestamp) noexcept
    : id(std::move(id))
    , file_id(std::move(file_id))
    , state(state)
    , alerts()
    , receivers()
    , timestamp(timestamp) {
}
 
Status::Status(std::string id, std::string file_id) noexcept
    : id(std::move(id))
    , file_id(std::move(file_id))
    , state(State::NotStarted)
    , alerts()
    , receivers()
    , timestamp(TimePoint::clock::now()) {
}
 
bool Status::operator==(Status const& rhs) const noexcept {
    return id == rhs.id && file_id == rhs.file_id && state == rhs.state && alerts == rhs.alerts &&
           receivers == rhs.receivers && result == rhs.result;
}
 
bool Status::operator!=(Status const& rhs) const noexcept {
    return !(*this == rhs);
}
 
bool HasError(Status const& status) noexcept {
    return !status.alerts.empty();
}
 
std::ostream& operator<<(std::ostream& os, Status const& s) {
    os << "Status(id='" << s.id << "', file_id='" << s.file_id << "', state=" << s.state
       << ", error=" << (HasError(s) ? "true" : "false") << ", result='" << s.result
       << "', timestamp=" << PutTime(s.timestamp) << ")";
    return os;
}
 
ObservableStatus::DeferSignal::DeferSignal(ObservableStatus* status, bool force)
    : m_status(status), m_force(force) {
    if (m_status) {
        m_status->DisableSignals();
    }
}
 
ObservableStatus::DeferSignal::DeferSignal(ObservableStatus::DeferSignal&& other) noexcept
    : m_status(nullptr) {
    std::swap(m_status, other.m_status);
    std::swap(m_force, other.m_force);
}
 
ObservableStatus::DeferSignal&
ObservableStatus::DeferSignal::operator=(ObservableStatus::DeferSignal&& other) noexcept {
    // Reset this before stealing resources from other
    Reset();
 
    std::swap(m_status, other.m_status);
    std::swap(m_force, other.m_force);
    return *this;
}
 
ObservableStatus::DeferSignal::~DeferSignal() noexcept {
    Reset();
}
 
void ObservableStatus::DeferSignal::Reset() noexcept {
    if (!m_status) {
        return;
    }
    m_status->EnableSignals(m_force);
    m_status = nullptr;
}
 
ObservableStatus::AlertActivator::AlertActivator(ObservableStatus* status, AlertId id)
    : m_status(status), m_defer(status), m_id(std::move(id)) {
}
 
ObservableStatus::AlertActivator::AlertActivator(ObservableStatus* status,
                                                 std::string_view category,
                                                 std::string key)
    : m_status(status), m_defer(status), m_id{std::string(category), std::move(key)} {
}
 
void ObservableStatus::AlertActivator::Set(std::string description) {
    BOOST_ASSERT_MSG(m_status != nullptr, "ObservableStatus is a nullptr");
    m_status->SetAlert(MakeAlert(m_id, std::move(description)));
}
 
void ObservableStatus::AlertActivator::Clear() {
    BOOST_ASSERT_MSG(m_status != nullptr, "ObservableStatus is a nullptr");
    m_status->ClearAlert(m_id);
}
 
ObservableStatus::ObservableStatus(Status status) : m_status(std::move(status)) {
}
 
ObservableStatus::ObservableStatus(std::string id, std::string file_id) noexcept
    : m_status(std::move(id), std::move(file_id)) {
}
 
ObservableStatus& ObservableStatus::operator=(Status const& status) {
    if (m_status == status && m_status.timestamp == status.timestamp) {
        return *this;
    }
    if (m_status.id != status.id) {
        throw std::invalid_argument(fmt::format(
            "Precondition not met (equality of DAQ IDs): {} != {}", m_status.id, status.id));
    }
    m_status = status;
    RecordChanges(true, UpdateTimestamp::No);
    return *this;
}
 
bool ObservableStatus::operator==(ObservableStatus const& rhs) const noexcept {
    return m_status == rhs.m_status;
}
 
bool ObservableStatus::operator!=(ObservableStatus const& rhs) const noexcept {
    return !(*this == rhs);
}
 
bool ObservableStatus::operator==(Status const& rhs) const noexcept {
    return m_status == rhs;
}
 
bool ObservableStatus::operator!=(Status const& rhs) const noexcept {
    return !(*this == rhs);
}
 
std::string const& ObservableStatus::GetId() const noexcept {
    return m_status.id;
}
 
std::vector<Alert> const& ObservableStatus::GetAlerts() const noexcept {
    return m_status.alerts;
}
 
Status::Clock::time_point ObservableStatus::GetTimestamp() const noexcept {
    return m_status.timestamp;
}
 
std::string const& ObservableStatus::GetFileId() const noexcept {
    return m_status.file_id;
}
 
State ObservableStatus::GetState() const noexcept {
    return m_status.state;
}
 
bool ObservableStatus::HasError() const noexcept {
    return daq::HasError(m_status);
}
 
void ObservableStatus::SetState(State s) noexcept {
    if (s == m_status.state) {
        return;
    }
    m_status.state = s;
    RecordChanges();
}
 
void ObservableStatus::SetReceiverStatus(std::size_t index, ReceiverStatus status) {
    auto [it, inserted] = m_status.receivers.try_emplace(index, status);
    if (inserted) {
        RecordChanges();
    } else if (it->second != status) {
        it->second = status;
        RecordChanges();
    }
}
 
ReceiverStatus ObservableStatus::GetReceiverStatus(std::size_t index) const noexcept {
    if (auto it = m_status.receivers.find(index); it != m_status.receivers.end()) {
        return it->second;
    } else {
        return ReceiverStatus{ReceiverStatus::State::NotStarted};
    }
}
 
void ObservableStatus::SetResult(std::string result) {
    m_status.result = std::move(result);
    RecordChanges();
}
 
void ObservableStatus::SetAlert(Alert alert) {
    ::daq::SetAlert(m_status.alerts, std::move(alert));
    RecordChanges();
}
 
void ObservableStatus::ClearAlert(AlertId const& alert) {
    if (::daq::ClearAlert(m_status.alerts, alert)) {
        RecordChanges();
    }
}
 
void ObservableStatus::EmitSignal(bool force) noexcept {
    // Unless forced, emit signal if not blocked and there are changes.
    if (force || (m_deferred_signals == 0 && m_changes_since_signal > 0u)) {
        try {
            m_signal(*this);
        } catch (...) {
        }
        // Reset number of changes since emitted signal
        m_changes_since_signal = 0u;
    }
}
 
void ObservableStatus::RecordChanges(bool force, UpdateTimestamp update) noexcept {
    m_changes_since_signal++;
    if (update == UpdateTimestamp::Yes) {
        m_status.timestamp = Status::TimePoint::clock::now();
    }
    EmitSignal(force);
}
 
ObservableStatus::operator Status() const {
    return m_status;
}
 
Status const& ObservableStatus::GetStatus() const noexcept {
    return m_status;
}
 
void ObservableStatus::EnableSignals(bool force) noexcept {
    assert(m_deferred_signals > 0);
    --m_deferred_signals;
 
    EmitSignal(force);
}
 
void ObservableStatus::DisableSignals() noexcept {
    ++m_deferred_signals;
}
 
std::ostream& operator<<(std::ostream& os, ObservableStatus const& s) {
    os << "ObservableStatus(id='" << s.GetId() << "', file_id='" << s.GetFileId()
       << "', state=" << s.GetState() << ", error=" << (s.HasError() ? "true" : "false") << ")";
    return os;
}
 
}  // namespace daq