ddsPubThread.hpp

Go to the documentation of this file.

 
#ifndef RTCTK_REUSABLECOMPONENT_TELREPUB_DDS_PUB_THREAD_HPP
#define RTCTK_REUSABLECOMPONENT_TELREPUB_DDS_PUB_THREAD_HPP
 
#include "agnostictopicif.h"
#include "mudpiProcessor.hpp"
#include "queue.hpp"
#include <cstdint>
#include <fastdds/dds/publisher/DataWriter.hpp>
#include <fmt/core.h>
#include <fmt/printf.h>
#include <memory>
#include <rtctk/componentFramework/bufferManager.hpp>
#include <rtctk/componentFramework/componentMetricsIf.hpp>
#include <rtctk/componentFramework/ddsCommon.hpp>
#include <rtctk/componentFramework/durationMonitor.hpp>
#include <rtctk/componentFramework/logger.hpp>
#include <rtctk/telRepub/wrangler.hpp>
#include <tbb/concurrent_queue.h>
 
#include <numapp/thread.hpp>
 
#include <chrono>
#include <iostream>
#include <utility>
 
namespace rtctk::telRepub {
 
using namespace rtctk::componentFramework;
 
class PubThread {
public:
    PubThread(std::uint16_t idx,
              DataWriter &data_writer,
              componentFramework::ComponentMetricsIf &metrics,
              int32_t topic_id,
              std::uint32_t queue_size,
              numapp::NumaPolicies const &thread_policies,
              std::uint16_t sim_freq,
              std::uint32_t sim_payload_size,
              std::uint32_t expected_sample_id_increment,
              WranglerFunction &&wrangler);
 
    ~PubThread();
 
    void RunAsync();
 
    void IdleAsync();
 
    bool InSimMode();
 
    MudpiProcessor &GetMudpiProcessor();
 
    std::string GetTopicName();
 
    std::shared_ptr<Queue> GetQueue();
 
private:
    void ProcessingLoop();
 
    void SimProcessingLoop();
 
    std::string MakeThreadDpName(bool is_sim_thread = true);
    std::string MakeThreadName(bool is_sim_thread = false);
 
    enum class State { Idle, Run, Exit };
 
    std::shared_ptr<Queue> m_frame_queue;
    std::uint16_t m_idx;
    std::string m_topic_name;
 
    log4cplus::Logger m_logger;
 
    std::atomic<State> m_requested_state;
    std::thread m_thread;
    std::string m_thread_name;
 
    MudpiProcessor m_mudpi_processor;
 
    AgnosticTopic m_sim_msg;
    DataWriter &m_data_writer;
 
    ComponentMetricsIf &m_metrics;
 
    std::uint16_t m_sim_freq;
    std::uint32_t m_sim_payload_size;              // bytes
    std::uint32_t m_expected_sample_id_increment;  // used in sim mode only
    // @note: OLDB does not support uint64_t which we really want, so we have to use signed 64.
    perfc::CounterI64 m_pc_published_samples;
    perfc::ScopedRegistration m_pc_published_samples_reg;
 
    perfc::CounterI64 m_pc_publication_errors;
    perfc::ScopedRegistration m_pc_publication_errors_reg;
 
    perfc::CounterI64 m_pc_last_sample_id_published;
    perfc::ScopedRegistration m_pc_last_sample_id_published_reg;
 
    std::unique_ptr<DurationMonitor<>> m_dur_mon;
 
    WranglerFunction m_wrangler;
};
 
}  // namespace rtctk::telRepub
 
#endif  // RTCTK_REUSABLECOMPONENT_TELREPUB_DDS_PUB_THREAD_HPP