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.h"
#include "mudpiProcessingError.hpp"
#include "queue.hpp"
#include <gsl/span>
#include <rtctk/componentFramework/componentMetricsIf.hpp>
#include <rtctk/componentFramework/frequencyEstimator.hpp>
#include <vector>
 
#include "mudpi.h"
#include "rtctk/componentFramework/durationMonitor.hpp"
#include "rtctk/telRepub/wrangler.hpp"
 
namespace rtctk::telRepub {
 
uint32_t const MAX_TOPIC_SIZE = 2560000;  //  Large increase for Pixel topics
 
class MudpiProcessor {
    using AgnosticTopic = rtctk::componentFramework::AgnosticTopic;
    using ComponentMetricsIf = rtctk::componentFramework::ComponentMetricsIf;
    struct PacketData {
        std::vector<QueueElement> queue_elements{};
        std::vector<gsl::span<const uint8_t>> payloads{};
        double time_stamp{0};
        uint32_t sample_id{0};
    };
 
    QueuePtr m_queue;
    PacketData m_packet_data{};
 
    ComponentMetricsIf& m_metrics;
 
    uint16_t m_num_frames = -1;  // TBC !!!
    mudpi::int32_t m_topic_id = 0;
 
    std::string m_topic_name;
    std::string m_metric_path_prefix;
 
    uint32_t m_expected_sample_id = 0;
    uint16_t m_expected_frame_id = 0;
 
    uint32_t m_expected_sample_id_increment;
 
    bool m_sample_finished = false;
 
    enum ProcessorState { 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;
 
public:
    explicit MudpiProcessor(mudpi::int32_t mudpi_topic_id,
                            std::string topic_name,
                            QueuePtr queue,
                            uint32_t expected_sample_id_increment,
                            componentFramework::ComponentMetricsIf& metrics);
 
    uint16_t GetTopicId() const;
 
    ErrorCode<MudpiProcessorError>
    ProcessMudpi(AgnosticTopic& topic, const WranglerFunction& wrangler);
 
    void ResetCounters();
 
};  // 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