eventRecordingUnit.hpp

Go to the documentation of this file.

 
#ifndef RTCTK_COMPONENTFRAMEWORK_EVENTRECORDINGUNIT_HPP
#define RTCTK_COMPONENTFRAMEWORK_EVENTRECORDINGUNIT_HPP
 
#include <rtctk/componentFramework/componentMetricsIf.hpp>
#include <rtctk/componentFramework/csvDataRecorder.hpp>
#include <rtctk/componentFramework/eventDefinitions.hpp>
#include <rtctk/componentFramework/eventServiceIf.hpp>
#include <rtctk/componentFramework/exceptions.hpp>
#include <rtctk/componentFramework/fitsDataRecorder.hpp>
#include <rtctk/componentFramework/recordingInfo.hpp>
#include <rtctk/componentFramework/recordingUnit.hpp>
#include <rtctk/componentFramework/typedEventService.hpp>
 
#include <fmt/format.h>
#include <numapp/numapolicies.hpp>
#include <numapp/thread.hpp>
#include <taiclock/taiClock.hpp>
 
#include <cstdint>
#include <exception>
#include <string>
#include <string_view>
#include <thread>
 
namespace rtctk::componentFramework {
 
template <>
struct [[deprecated]] RecordingInfo<ComputationEvent> {
    using sv = std::string_view;
    using DataType = ComputationEvent;
    using Recorder = CsvDataRecorder<taiclock::TaiClock::time_point::rep,
                                     std::string,
                                     std::string,
                                     std::string,
                                     std::string>;
 
    static constexpr Recorder::ColumnDescription COLUMNS =
        Recorder::ColumnDescription{{{sv("timestamp"), sv("")},
                                     {sv("source_timestamp"), sv("")},
                                     {sv("origin"), sv("")},
                                     {sv("item"), sv("")},
                                     {sv("state"), sv("")}}};
 
    static Recorder::TupleType AsTuple(const DataType& data) {
        auto ts = taiclock::TaiClock::now().time_since_epoch().count();
        return std::make_tuple(ts, data.time, data.origin, data.item, data.state);
    }
};
 
template <>
struct [[deprecated]] RecordingInfo<ComputationFinishedEvent> {
    using sv = std::string_view;
    using DataType = ComputationFinishedEvent;
    using Recorder = CsvDataRecorder<taiclock::TaiClock::time_point::rep,
                                     std::string,
                                     std::string,
                                     std::string,
                                     std::string>;
 
    static constexpr Recorder::ColumnDescription COLUMNS =
        Recorder::ColumnDescription{{{sv("timestamp"), sv("")},
                                     {sv("source_timestamp"), sv("")},
                                     {sv("origin"), sv("")},
                                     {sv("item"), sv("")},
                                     {sv("result"), sv("")}}};
 
    static Recorder::TupleType AsTuple(const DataType& data) {
        auto ts = taiclock::TaiClock::now().time_since_epoch().count();
        return std::make_tuple(ts, data.time, data.origin, data.item, data.result.dump());
    }
};
 
template <>
struct [[deprecated]] RecordingInfo<ComputationStartedEvent> {
    using sv = std::string_view;
    using DataType = ComputationStartedEvent;
    using Recorder =
        CsvDataRecorder<taiclock::TaiClock::time_point::rep, std::string, std::string, std::string>;
 
    static constexpr Recorder::ColumnDescription COLUMNS =
        Recorder::ColumnDescription{{{sv("timestamp"), sv("")},
                                     {sv("source_timestamp"), sv("")},
                                     {sv("origin"), sv("")},
                                     {sv("item"), sv("")}}};
 
    static Recorder::TupleType AsTuple(const DataType& data) {
        auto ts = taiclock::TaiClock::now().time_since_epoch().count();
        return std::make_tuple(ts, data.time, data.origin, data.item);
    }
};
 
template <>
struct [[deprecated]] RecordingInfo<ConfigurationEvent> {
    using sv = std::string_view;
    using DataType = ConfigurationEvent;
    using Recorder = CsvDataRecorder<taiclock::TaiClock::time_point::rep,
                                     std::string,
                                     std::string,
                                     std::string,
                                     std::string>;
 
    static constexpr Recorder::ColumnDescription COLUMNS =
        Recorder::ColumnDescription{{{sv("timestamp"), sv("")},
                                     {sv("source_timestamp"), sv("")},
                                     {sv("origin"), sv("")},
                                     {sv("items"), sv("")},
                                     {sv("state"), sv("")}}};
 
    static Recorder::TupleType AsTuple(const DataType& data) {
        auto ts = taiclock::TaiClock::now().time_since_epoch().count();
        std::string items;
        bool is_first = true;
        for (const auto& item : data.items) {
            if (is_first == false) {
                items += ";";
            }
            items += item.name;
            is_first = false;
        }
        return std::make_tuple(ts, data.time, data.origin, items, data.state);
    }
};
 
template <>
struct [[deprecated]] RecordingInfo<ConfigurationUpdatedEvent> {
    using sv = std::string_view;
    using DataType = ConfigurationUpdatedEvent;
    using Recorder = CsvDataRecorder<taiclock::TaiClock::time_point::rep,
                                     std::string,
                                     std::string,
                                     std::string,
                                     std::string>;
 
    static constexpr Recorder::ColumnDescription COLUMNS =
        Recorder::ColumnDescription{{{sv("timestamp"), sv("")},
                                     {sv("source_timestamp"), sv("")},
                                     {sv("origin"), sv("")},
                                     {sv("items"), sv("")},
                                     {sv("state"), sv("")}}};
 
    static Recorder::TupleType AsTuple(const DataType& data) {
        auto ts = taiclock::TaiClock::now().time_since_epoch().count();
        std::string items;
        bool is_first = true;
        for (const auto& item : data.items) {
            if (is_first == false) {
                items += ";";
            }
            items += item.name;
            is_first = false;
        }
        return std::make_tuple(ts, data.time, data.origin, items, data.state);
    }
};
 
template <>
struct [[deprecated]] RecordingInfo<ConfigurationRetrievedEvent> {
    using sv = std::string_view;
    using DataType = ConfigurationRetrievedEvent;
    using Recorder = CsvDataRecorder<taiclock::TaiClock::time_point::rep,
                                     std::string,
                                     std::string,
                                     std::string,
                                     std::string>;
 
    static constexpr Recorder::ColumnDescription COLUMNS =
        Recorder::ColumnDescription{{{sv("timestamp"), sv("")},
                                     {sv("source_timestamp"), sv("")},
                                     {sv("origin"), sv("")},
                                     {sv("items"), sv("")},
                                     {sv("state"), sv("")}}};
 
    static Recorder::TupleType AsTuple(const DataType& data) {
        auto ts = taiclock::TaiClock::now().time_since_epoch().count();
        std::string items;
        bool is_first = true;
        for (const auto& item : data.items) {
            if (is_first == false) {
                items += ";";
            }
            items += item.name;
            is_first = false;
        }
        return std::make_tuple(ts, data.time, data.origin, items, data.state);
    }
};
 
template <>
struct [[deprecated]] RecordingInfo<HrtcConfigurationScheduledEvent> {
    using sv = std::string_view;
    using DataType = HrtcConfigurationScheduledEvent;
    using Recorder = CsvDataRecorder<taiclock::TaiClock::time_point::rep,
                                     std::string,
                                     std::string,
                                     std::string,
                                     std::string,
                                     std::string,
                                     std::string,
                                     uint64_t>;
 
    static constexpr Recorder::ColumnDescription COLUMNS =
        Recorder::ColumnDescription{{{sv("timestamp"), sv("")},
                                     {sv("source_timestamp"), sv("")},
                                     {sv("origin"), sv("")},
                                     {sv("items"), sv("")},
                                     {sv("state"), sv("")},
                                     {sv("hrtc_origin"), sv("")},
                                     {sv("hrtc_timestamp"), sv("")},
                                     {sv("hrtc_sample_id"), sv("")}}};
 
    static Recorder::TupleType AsTuple(const DataType& data) {
        auto ts = taiclock::TaiClock::now().time_since_epoch().count();
        std::string items;
        bool is_first = true;
        for (const auto& item : data.items) {
            if (is_first == false) {
                items += ";";
            }
            items += item.name;
            is_first = false;
        }
        return std::make_tuple(ts,
                               data.time,
                               data.origin,
                               items,
                               data.state,
                               data.hrtc_origin,
                               data.hrtc_timestamp,
                               data.hrtc_sample_id);
    }
};
 
template <>
struct [[deprecated]] RecordingInfo<HrtcConfigurationAppliedEvent> {
    using sv = std::string_view;
    using DataType = HrtcConfigurationAppliedEvent;
    using Recorder = CsvDataRecorder<taiclock::TaiClock::time_point::rep,
                                     std::string,
                                     std::string,
                                     std::string,
                                     std::string,
                                     std::string,
                                     std::string,
                                     uint64_t>;
 
    static constexpr Recorder::ColumnDescription COLUMNS =
        Recorder::ColumnDescription{{{sv("timestamp"), sv("")},
                                     {sv("source_timestamp"), sv("")},
                                     {sv("origin"), sv("")},
                                     {sv("items"), sv("")},
                                     {sv("state"), sv("")},
                                     {sv("hrtc_origin"), sv("")},
                                     {sv("hrtc_timestamp"), sv("")},
                                     {sv("hrtc_sample_id"), sv("")}}};
 
    static Recorder::TupleType AsTuple(const DataType& data) {
        auto ts = taiclock::TaiClock::now().time_since_epoch().count();
        std::string items;
        bool is_first = true;
        for (const auto& item : data.items) {
            if (is_first == false) {
                items += ";";
            }
            items += item.name;
            is_first = false;
        }
        return std::make_tuple(ts,
                               data.time,
                               data.origin,
                               items,
                               data.state,
                               data.hrtc_origin,
                               data.hrtc_timestamp,
                               data.hrtc_sample_id);
    }
};
 
template <>
struct [[deprecated]] RecordingInfo<CoordinationEvent> {
    using sv = std::string_view;
    using DataType = CoordinationEvent;
    using Recorder = CsvDataRecorder<taiclock::TaiClock::time_point::rep, std::string, std::string>;
 
    static constexpr Recorder::ColumnDescription COLUMNS = Recorder::ColumnDescription{
        {{sv("timestamp"), sv("")}, {sv("source_timestamp"), sv("")}, {sv("origin"), sv("")}}};
 
    static Recorder::TupleType AsTuple(const DataType& data) {
        auto ts = taiclock::TaiClock::now().time_since_epoch().count();
        return std::make_tuple(ts, data.time, data.origin);
    }
};
 
template <>
struct [[deprecated]] RecordingInfo<StateChangedEvent> {
    using sv = std::string_view;
    using DataType = StateChangedEvent;
    using Recorder = CsvDataRecorder<taiclock::TaiClock::time_point::rep,
                                     std::string,
                                     std::string,
                                     std::string,
                                     std::string>;
 
    static constexpr Recorder::ColumnDescription COLUMNS =
        Recorder::ColumnDescription{{{sv("timestamp"), sv("")},
                                     {sv("source_timestamp"), sv("")},
                                     {sv("origin"), sv("")},
                                     {sv("entity"), sv("")},
                                     {sv("state"), sv("")}}};
 
    static Recorder::TupleType AsTuple(const DataType& data) {
        auto ts = taiclock::TaiClock::now().time_since_epoch().count();
        return std::make_tuple(ts, data.time, data.origin, data.entity, data.state);
    }
};
 
template <>
struct [[deprecated]] RecordingInfo<HrtcStateChangedEvent> {
    using sv = std::string_view;
    using DataType = HrtcStateChangedEvent;
    using Recorder = CsvDataRecorder<taiclock::TaiClock::time_point::rep,
                                     std::string,
                                     std::string,
                                     std::string,
                                     std::string,
                                     std::string,
                                     uint64_t>;
 
    static constexpr Recorder::ColumnDescription COLUMNS =
        Recorder::ColumnDescription{{{sv("timestamp"), sv("")},
                                     {sv("source_timestamp"), sv("")},
                                     {sv("origin"), sv("")},
                                     {sv("entity"), sv("")},
                                     {sv("state"), sv("")},
                                     {sv("hrtc_timestamp"), sv("")},
                                     {sv("hrtc_sample_id"), sv("")}}};
 
    static Recorder::TupleType AsTuple(const DataType& data) {
        auto ts = taiclock::TaiClock::now().time_since_epoch().count();
        return std::make_tuple(ts,
                               data.time,
                               data.origin,
                               data.entity,
                               data.state,
                               data.hrtc_timestamp,
                               data.hrtc_sample_id);
    }
};
 
template <>
struct [[deprecated]] RecordingInfo<AlertStatusEvent> {
    using sv = std::string_view;
    using DataType = AlertStatusEvent;
    using Recorder = CsvDataRecorder<taiclock::TaiClock::time_point::rep,
                                     std::string,
                                     std::string,
                                     bool,
                                     std::string>;
 
    static constexpr Recorder::ColumnDescription COLUMNS =
        Recorder::ColumnDescription{{{sv("timestamp"), sv("")},
                                     {sv("source_timestamp"), sv("")},
                                     {sv("origin"), sv("")},
                                     {sv("status"), sv("")},
                                     {sv("active_alerts"), sv("")}}};
 
    static Recorder::TupleType AsTuple(const DataType& data) {
        auto ts = taiclock::TaiClock::now().time_since_epoch().count();
 
        JsonPayload json_active;
        for (auto&& alert : data.active_alerts) {
            json_active.push_back(JsonPayload{{"origin", alert.origin},
                                              {"id", alert.id},
                                              {"description", alert.description},
                                              {"since", alert.since}});
        }
 
        return std::make_tuple(
            ts, data.time, data.origin, data.status, json_active.dump());
    }
};
 
template <typename EventType, typename OutputStage = typename RecordingInfo<EventType>::Recorder>
class [[deprecated("Use EventRecordingUnit")]] TypedEventRecordingUnit : public RecordingUnit {
public:
    using FilterMethod = std::function<bool(const EventType&)>;
 
    TypedEventRecordingUnit(const std::string& comp_id,
                            const std::string& unit_id,
                            ServiceContainer& services,
                            FilterMethod input_filter = nullptr)
        : RecordingUnit(comp_id, unit_id, "TypedEvent", services)
        , m_event_service(services.Get<EventServiceIf>())
        , m_subscriber(m_event_service.MakeSubscriber<EventType>())
        , m_input_filter(input_filter)
        , m_output(RecordingInfo<EventType>::COLUMNS) {
        auto topic_path = DataPointPath(fmt::format(OLDB_PATH_TOPIC_NAME, comp_id, unit_id));
        m_oldb.CreateDataPoint<std::string>(topic_path);
        m_oldb.SetDataPoint<std::string>(topic_path, EventType::TOPIC_NAME);
 
        m_samples_written_reg = m_metrics.AddCounter(
            &m_samples_written, CounterMetricInfo(unit_id + "/samples_written", "Samples written"));
        Update();
    }
 
    virtual ~TypedEventRecordingUnit() {
        m_stop = true;
        if (m_process_thread.joinable()) {
            m_process_thread.join();
        }
        SetStopped();
    }
 
    void Prepare(const std::filesystem::path& file_path) override {
        SetState(
            State::PREPARING, State::STOPPED, "tried to prepare a non-stopped EventRecordingUnit");
 
        m_file_path = file_path / (GetId() + m_output.DefaultFileExtension());
 
        auto policies = numapp::NumaPolicies();
        m_process_thread =
            numapp::MakeThread(m_unit_id.substr(0, 15), policies, [&]() { return Process(); });
    }
    void Start() override {
        m_start = true;
    }
    std::vector<std::filesystem::path> Stop() override {
        m_stop = true;
        if (m_process_thread.joinable()) {
            m_process_thread.join();
        }
 
        std::vector<std::filesystem::path> files;
        if (m_file_path) {
            files.push_back(*m_file_path);
        }
        m_file_path.reset();
        SetStopped();
        return files;
    }
 
private:
    void Process() {
        using namespace std::chrono_literals;
 
        try {
            m_output.Open(*m_file_path);
            m_samples_written.Store(0);
 
            m_subscriber->Subscribe([&](const EventType& ev) {
                if (GetState() == State::RUNNING) {
                    try {
                        if (m_input_filter) {
                            if (m_input_filter(ev)) {
                                m_output.Write(RecordingInfo<EventType>::AsTuple(ev));
                                m_samples_written++;
                            }
                        } else {
                            m_output.Write(RecordingInfo<EventType>::AsTuple(ev));
                            m_samples_written++;
                        }
                    } catch (...) {
                        SetFailed(std::current_exception());
                        return;
                    }
                }
            });
 
            SetState(State::IDLE,
                     State::PREPARING,
                     "EventRecordingUnit tried to go to IDLE from another state than PREPARING");
 
            while (m_stop == false) {
                switch (GetState()) {
                case State::IDLE:
                    if (m_start) {
                        SetState(State::WAITING,
                                 State::IDLE,
                                 "EventRecorder illegal state transition, tried going to WAITING "
                                 "from another state than IDLE");
                        break;
                    }
                    std::this_thread::sleep_for(1ms);
                    break;
                case State::WAITING:
                    if (not HasLeaders() or (HasLeaders() and HasFirstLeaderStarted())) {
                        SetState(State::RUNNING,
                                 State::WAITING,
                                 "EventRecorder illegal state transition, tried going to RUNNING "
                                 "from another state than WAITING");
                        break;
                    }
                    std::this_thread::sleep_for(1ms);
                    break;
                case State::RUNNING:
                    if (HasLeaders() and HasLastLeaderFinished()) {
                        SetState(State::FINISHED,
                                 State::RUNNING,
                                 "EventRecorder illegal state transition, tried going to FINISHED "
                                 "from another state than RUNNING");
                        break;
                    }
                    std::this_thread::sleep_for(1ms);
                    break;
                default:
                    std::this_thread::sleep_for(1ms);
                }
            }
 
            m_subscriber->Unsubscribe();
            m_output.Close();
            // Stopped state will be set by Stop function after join
            m_start = false;
            m_stop = false;
            ResetLeaderStates();
        } catch (...) {
            m_subscriber->Unsubscribe();
            m_output.Close();
            m_start = false;
            m_stop = false;
            SetFailed(std::current_exception());
            ResetLeaderStates();
        }
    }
 
// we dont need any deprecation warnings in already deprecated code
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wdeprecated-declarations"
    TypedEventService m_event_service;
    std::unique_ptr<TypedEventSubscriber<EventType>> m_subscriber;
#pragma GCC diagnostic pop
 
    FilterMethod m_input_filter;
 
    OutputStage m_output;
 
    std::atomic<bool> m_start = false;
    std::atomic<bool> m_stop = false;
    std::thread m_process_thread;
 
    perfc::CounterI64 m_samples_written;
    perfc::ScopedRegistration m_samples_written_reg;
 
    inline static constexpr std::string_view OLDB_PATH_TOPIC_NAME = "/{}/rec_units/{}/topic_name";
};
 
template <typename OutputStage = CsvDataRecorder<taiclock::TaiClock::time_point::rep, std::string>>
class [[deprecated("Use EventRecordingUnit")]] JsonEventRecordingUnit : public RecordingUnit {
public:
    using sv = std::string_view;
    using FilterMethod = std::function<bool(const JsonPayload&)>;
    using ColumnDescription = typename OutputStage::ColumnDescription;
    using ColumnMetaData = typename OutputStage::ColumnMetaData;
 
    JsonEventRecordingUnit(const std::string& comp_id,
                           const std::string& unit_id,
                           ServiceContainer& services,
                           const std::string& topic_name,
                           FilterMethod input_filter = nullptr)
        : RecordingUnit(comp_id, unit_id, "JsonEvent", services)
        , m_event_service(services.Get<EventServiceIf>())
        , m_subscriber(m_event_service.MakeSubscriber(topic_name))
        , m_input_filter(std::move(input_filter))
        , m_output(COLUMNS) {
        auto topic_path = DataPointPath(fmt::format(OLDB_PATH_TOPIC_NAME, comp_id, unit_id));
        m_oldb.CreateDataPoint<std::string>(topic_path);
        m_oldb.SetDataPoint<std::string>(topic_path, topic_name);
 
        m_samples_written_reg = m_metrics.AddCounter(
            &m_samples_written, CounterMetricInfo(unit_id + "/samples_written", "Samples written"));
        Update();
    }
 
    virtual ~JsonEventRecordingUnit() {
        m_stop = true;
        if (m_process_thread.joinable()) {
            m_process_thread.join();
        }
        SetStopped();
    }
 
    void Prepare(const std::filesystem::path& file_path) override {
        SetState(State::PREPARING,
                 State::STOPPED,
                 "EventRecorder tried to go from non-STOPPED State to PREPARING");
 
        m_file_path = file_path / (GetId() + m_output.DefaultFileExtension());
 
        auto policies = numapp::NumaPolicies();
        m_process_thread =
            numapp::MakeThread(m_unit_id.substr(0, 15), policies, [&]() { return Process(); });
    }
    void Start() override {
        m_start = true;
    }
    std::vector<std::filesystem::path> Stop() override {
        m_stop = true;
        if (m_process_thread.joinable()) {
            m_process_thread.join();
        }
        SetStopped();
        std::vector<std::filesystem::path> files;
        if (m_file_path) {
            files.push_back(*m_file_path);
        }
        return files;
    }
 
private:
    void Process() {
        using namespace std::chrono_literals;
 
        try {
            m_output.Open(*m_file_path);
            m_samples_written.Store(0);
 
            m_subscriber->Subscribe([&](const JsonPayload& ev) {
                if (GetState() == State::RUNNING) {
                    try {
                        if (m_input_filter) {
                            if (m_input_filter(ev)) {
                                m_output.Write(AsTuple(ev));
                                m_samples_written++;
                            }
                        } else {
                            m_output.Write(AsTuple(ev));
                            m_samples_written++;
                        }
                    } catch (...) {
                        SetFailed(std::current_exception());
                        return;
                    }
                }
            });
 
            SetState(State::IDLE,
                     State::PREPARING,
                     "Event recorder was not in PREPARING when trying to go to IDLE");
 
            while (m_stop == false) {
                switch (GetState()) {
                case State::IDLE:
                    if (m_start) {
                        SetState(State::WAITING,
                                 State::IDLE,
                                 "Event recorder was not in IDLE when trying to go to WAITING");
                        break;
                    }
                    std::this_thread::sleep_for(1ms);
                    break;
                case State::WAITING:
                    if (not HasLeaders() or (HasLeaders() and HasFirstLeaderStarted())) {
                        SetState(State::RUNNING,
                                 State::WAITING,
                                 "Event recorder was not in WAITING when trying to go to RUNNING");
                        break;
                    }
                    std::this_thread::sleep_for(1ms);
                    break;
                case State::RUNNING:
                    if (HasLeaders() and HasLastLeaderFinished()) {
                        SetState(State::FINISHED,
                                 State::RUNNING,
                                 "Event recorder was not running when trying to go to finish");
                        break;
                    }
                    std::this_thread::sleep_for(1ms);
                    break;
                default:
                    std::this_thread::sleep_for(1ms);
                }
            }
 
            m_subscriber->Unsubscribe();
            m_output.Close();
            // the actual STOPPED state will be set by the Stop function
            m_start = false;
            m_stop = false;
            ResetLeaderStates();
        } catch (...) {
            m_subscriber->Unsubscribe();
            m_output.Close();
            m_start = false;
            m_stop = false;
            SetFailed(std::current_exception());
            ResetLeaderStates();
        }
    }
 
    static typename OutputStage::TupleType AsTuple(const JsonPayload& data) {
        auto ts = taiclock::TaiClock::now().time_since_epoch().count();
        return std::make_tuple(ts, data.dump());
    }
 
    static constexpr ColumnDescription COLUMNS{ColumnMetaData{sv("timestamp"), sv("")},
                                               ColumnMetaData{sv("payload"), sv("")}};
 
    EventServiceIf& m_event_service;
    std::unique_ptr<EventSubscriberIf> m_subscriber;
 
    FilterMethod m_input_filter;
 
    OutputStage m_output;
 
    std::atomic<bool> m_start = false;
    std::atomic<bool> m_stop = false;
    std::thread m_process_thread;
 
    perfc::CounterI64 m_samples_written;
    perfc::ScopedRegistration m_samples_written_reg;
 
    inline static constexpr std::string_view OLDB_PATH_TOPIC_NAME = "/{}/rec_units/{}/topic_name";
};
 
template <class OutputStage = std::ofstream>
class EventRecordingUnit : public RecordingUnit {
public:
    using FilterMethod = std::function<bool(const JsonPayload&)>;
 
    EventRecordingUnit(const std::string& comp_id,
                       const std::string& unit_id,
                       ServiceContainer& services,
                       FilterMethod input_filter = nullptr)
        : RecordingUnit(comp_id, unit_id, "Event", services)
        , m_event_service(services.Get<EventServiceIf>()) {
        if (input_filter) {
            m_input_filter = std::move(input_filter);
        } else {
            m_input_filter = [](const JsonPayload&) { return true; };
        }
 
        auto rtr_topic_path = DataPointPath(fmt::format(RTR_PATH_TOPIC_NAME, comp_id, unit_id));
        auto topic_name = m_rtr.GetDataPoint<std::string>(rtr_topic_path);
 
        auto oldb_topic_path = DataPointPath(fmt::format(OLDB_PATH_TOPIC_NAME, comp_id, unit_id));
        m_oldb.CreateDataPoint<std::string>(oldb_topic_path);
        m_oldb.SetDataPoint<std::string>(oldb_topic_path, topic_name);
 
        m_subscriber = m_event_service.MakeSubscriber(topic_name);
 
        m_samples_written_reg = m_metrics.AddCounter(
            &m_samples_written, CounterMetricInfo(unit_id + "/samples_written", "Samples written"));
        Update();
    }
 
    virtual ~EventRecordingUnit() {
        m_stop = true;
        if (m_process_thread.joinable()) {
            m_process_thread.join();
        }
        SetStopped();
    }
 
    void Prepare(const std::filesystem::path& file_path) override {
        SetState(State::PREPARING,
                 State::STOPPED,
                 "EventRecorder tried to go from non-STOPPED State to PREPARING");
 
        m_file_path = file_path / (GetId() + ".jsonl");
 
        auto policies = numapp::NumaPolicies();
        m_process_thread =
            numapp::MakeThread(m_unit_id.substr(0, 15), policies, [&]() { return Process(); });
    }
    void Start() override {
        m_start = true;
    }
    std::vector<std::filesystem::path> Stop() override {
        m_stop = true;
        if (m_process_thread.joinable()) {
            m_process_thread.join();
        }
        SetStopped();
        std::vector<std::filesystem::path> files;
        if (m_file_path) {
            files.push_back(*m_file_path);
        }
        return files;
    }
 
private:
    void Process() {
        using namespace std::chrono_literals;
 
        OutputStage output;
 
        try {
            output.open(*m_file_path, std::ios::out);
            m_samples_written.Store(0);
 
            m_subscriber->Subscribe([&](const JsonPayload& ev) {
                if (GetState() == State::RUNNING) {
                    try {
                        if (m_input_filter(ev)) {
                            output << ev << "\n";
                            m_samples_written++;
                        }
                    } catch (...) {
                        SetFailed(std::current_exception());
                        return;
                    }
                }
            });
 
            SetState(State::IDLE,
                     State::PREPARING,
                     "Event recorder was not in PREPARING when trying to go to IDLE");
 
            while (m_stop == false) {
                switch (GetState()) {
                case State::IDLE:
                    if (m_start) {
                        SetState(State::WAITING,
                                 State::IDLE,
                                 "Event recorder was not in IDLE when trying to go to WAITING");
                        break;
                    }
                    std::this_thread::sleep_for(1ms);
                    break;
                case State::WAITING:
                    if (not HasLeaders() or (HasLeaders() and HasFirstLeaderStarted())) {
                        SetState(State::RUNNING,
                                 State::WAITING,
                                 "Event recorder was not in WAITING when trying to go to RUNNING");
                        break;
                    }
                    std::this_thread::sleep_for(1ms);
                    break;
                case State::RUNNING:
                    if (HasLeaders() and HasLastLeaderFinished()) {
                        SetState(State::FINISHED,
                                 State::RUNNING,
                                 "Event recorder was not running when trying to go to finish");
                        break;
                    }
                    std::this_thread::sleep_for(1ms);
                    break;
                default:
                    std::this_thread::sleep_for(1ms);
                }
            }
 
            m_subscriber->Unsubscribe();
            // the actual STOPPED state will be set by the Stop function
            m_start = false;
            m_stop = false;
            ResetLeaderStates();
        } catch (...) {
            m_subscriber->Unsubscribe();
            m_start = false;
            m_stop = false;
            SetFailed(std::current_exception());
            ResetLeaderStates();
        }
    }
 
    EventServiceIf& m_event_service;
    std::unique_ptr<EventSubscriberIf> m_subscriber;
 
    FilterMethod m_input_filter;
 
    std::atomic<bool> m_start = false;
    std::atomic<bool> m_stop = false;
    std::thread m_process_thread;
 
    perfc::CounterI64 m_samples_written;
    perfc::ScopedRegistration m_samples_written_reg;
 
    inline static constexpr std::string_view RTR_PATH_TOPIC_NAME =
        "/{}/static/rec_units/{}/topic_name";
 
    inline static constexpr std::string_view OLDB_PATH_TOPIC_NAME = "/{}/rec_units/{}/topic_name";
};
 
}  // namespace rtctk::componentFramework
 
#endif  // RTCTK_COMPONENTFRAMEWORK_EVENTRECORDINGUNIT_HPP