mudpiProcessor.hpp

Go to the documentation of this file.

#ifndef RTCTK_REUSABLECOMPONENT_TELREPUB_MUDPIPROCESSOR_HPP
#define RTCTK_REUSABLECOMPONENT_TELREPUB_MUDPIPROCESSOR_HPP
 
#include <boost/asio.hpp>
#include <cstdint>
#include <fmt/format.h>
 
#include "agnostictopicif.hpp"
#include "llnetio/mudpi/mudpi.hpp"
#include "mudpiProcessingError.hpp"
#include "queue.hpp"
#include "rtctk/componentFramework/durationMonitor.hpp"
#include "rtctk/telRepub/wrangler.hpp"
#include <gsl/span>
#include <limits>
#include <rtctk/componentFramework/alertServiceIf.hpp>
#include <rtctk/componentFramework/componentMetricsIf.hpp>
#include <rtctk/componentFramework/frequencyEstimator.hpp>
#include <rtctk/componentFramework/serviceContainer.hpp>
#include <taiclock/taiClock.hpp>
#include <vector>
 
namespace rtctk::telRepub {

struct CfgMudpiProc {
    std::string dds_topic_name;
    llnetio::mudpi::TopicId mudpi_topic_id;
    std::uint32_t expected_sample_id_increment = 1u;
    std::chrono::milliseconds receive_sample_timeout_ms =
        std::chrono::milliseconds(20000);  // default 20s
};

class MudpiProcessor {
    using AgnosticTopic = rtctk::componentFramework::AgnosticTopic;
    using ComponentMetricsIf = rtctk::componentFramework::ComponentMetricsIf;
    using AlertServiceIf = rtctk::componentFramework::AlertServiceIf;
    struct PacketData {
        std::vector<QueueElement> queue_elements{};
        std::vector<gsl::span<const uint8_t>> payloads{};
        taiclock::TaiClock::time_point time_stamp{};
        llnetio::mudpi::SampleId sample_id{0};
 
        void Reset() {
            queue_elements.clear();
            payloads.clear();
            time_stamp = {};
            sample_id = 0;
        }
    };
 
    QueuePtr m_queue;
    PacketData m_packet_data{};
 
    ComponentMetricsIf& m_metrics;
    AlertServiceIf& m_alerts;
 
    llnetio::mudpi::NumFrames m_num_frames = std::numeric_limits<llnetio::mudpi::NumFrames>::max();
    llnetio::mudpi::TopicId m_topic_id = 0;
 
    std::string m_topic_name;
    std::string m_metric_path_prefix;
 
    llnetio::mudpi::SampleId m_expected_sample_id = 0;
    llnetio::mudpi::FrameId m_expected_frame_id = 0;
 
    llnetio::mudpi::SampleId m_expected_sample_id_increment;
 
    bool m_sample_finished = false;
 
    enum ProcessorState : uint8_t { Synchronising, Processing };
 
    ProcessorState m_state = Synchronising;

    log4cplus::Logger m_logger;
 
    ErrorCode<MudpiProcessorError> StateMachine(PacketFrame packet, AgnosticTopic& topic);

    // @note: OLDB does not support uint64_t which we really want, so we have to use signed 64.
    perfc::CounterI64 m_pc_frames_recvd;
    perfc::ScopedRegistration m_pc_frames_recvd_reg;
    perfc::CounterI64 m_pc_samples_recvd;
    perfc::ScopedRegistration m_pc_samples_recvd_reg;
    perfc::CounterI64 m_pc_frame_errors;
    perfc::ScopedRegistration m_pc_frame_errors_reg;
    perfc::CounterI64 m_pc_sample_errors;
    perfc::ScopedRegistration m_pc_sample_errors_reg;
    perfc::CounterI64 m_pc_last_sample_id_recvd;
    perfc::ScopedRegistration m_pc_last_sample_id_recvd_reg;
 
    componentFramework::FrequencyEstimator<> m_freq_estimator;
    componentFramework::DurationMonitor<> m_wrangler_duration;
 
    std::atomic<std::chrono::steady_clock::time_point> m_last_sample_received;
 
    // receive timeout alert
    componentFramework::AlertSource m_receive_timeout_alert;
    std::chrono::milliseconds m_receive_sample_timeout_ms =
        std::chrono::milliseconds(0);  // default - we do not check for timeout
 
public:
    explicit MudpiProcessor(QueuePtr queue,
                            CfgMudpiProc& cfg,
                            componentFramework::ServiceContainer& service);

    uint16_t GetTopicId() const;

    ErrorCode<MudpiProcessorError>
    ProcessMudpi(AgnosticTopic& topic, const WranglerFunction& wrangler);

    void ResetCounters();

    bool Monitor();
 
};  // MudpiProcessor
 
std::error_code
DefaultWrangler(gsl::span<const gsl::span<const uint8_t>> input, std::vector<uint8_t>& output);
 
}  // namespace rtctk::telRepub
 
#endif  // RTCTK_REUSABLECOMPONENT_TELREPUB_MUDPIPROCESSOR_HPP