telRepubBusinessLogic.hpp

Go to the documentation of this file.

 
#ifndef RTCTK_REUSABLECOMPONENT_TELREPUB_TELREPUBBUSINESSLOGIC_HPP
#define RTCTK_REUSABLECOMPONENT_TELREPUB_TELREPUBBUSINESSLOGIC_HPP
 
#include <fastdds/dds/publisher/DataWriter.hpp>
#include <rtctk/componentFramework/ddsPub.hpp>
#include <rtctk/telRepub/wrangler.hpp>
 
#include "numapp/numapolicies.hpp"
#include "rtctk/componentFramework/componentMetricsIf.hpp"
#include "rtctk/componentFramework/logger.hpp"
#include "rtctk/componentFramework/rtcComponent.hpp"
#include "rtctk/componentFramework/runnable.hpp"
#include "rtctk/componentFramework/serviceContainer.hpp"
 
#include <rad/logger.hpp>
#include <rad/mal/utils.hpp>
 
#include <cstdint>
 
namespace rtctk::telRepub {
 
using namespace rtctk::componentFramework;
 
using LifeCycle = Runnable<RtcComponent>;
 
using RepubWranglerFactory = std::function<WranglerFunction(int32_t)>;
 
class PubThread;
 
class MudpiProcessor;
 
class UdpReceiver;
 
class TelRepubBusinessLogic : public LifeCycle::BizLogicIf {
public:
    using ComponentType = LifeCycle;
    TelRepubBusinessLogic(
        std::string name, ServiceContainer& services, RepubWranglerFactory&& factory = [](auto x) {
            return nullptr;
        });
 
    virtual ~TelRepubBusinessLogic();
 
    void ActivityStarting(StopToken st) override;
 
    void ActivityRecovering(StopToken st) override;
 
    void ActivityInitialising(StopToken st) override;
 
    void ActivityEnabling(StopToken st) override;
 
    void ActivityDisabling(StopToken st) override;
 
    void ActivityGoingRunning(StopToken st) override;
 
    void ActivityGoingIdle(StopToken st) override;
 
    void ActivityRunning(StopToken st) override;
 
    void ActivityUpdating(StopToken st, JsonPayload const& args) override;
 
    bool GuardUpdatingAllowed(JsonPayload const& args) override;
 
protected:
    struct DdsPublisherContext {
        DdsPublisherContext(std::uint16_t idx,
                            std::uint16_t sim_freq_arg,
                            std::uint32_t sim_payload_size_arg,
                            std::int32_t mudpi_topic_id_arg,
                            std::int32_t receiver_index_arg,
                            componentFramework::DataWriter& data_writer,
                            componentFramework::ComponentMetricsIf& metrics,
                            std::uint32_t queue_size,
                            std::uint32_t expected_sample_id_increment,
                            numapp::NumaPolicies const& thread_policies,
                            WranglerFunction&& wrangler);
        DdsPublisherContext(DdsPublisherContext&&) noexcept = default;
        std::uint16_t mudpi_topic_id;
        std::uint16_t receiver_index;
        std::unique_ptr<PubThread> pub_thread;
        WranglerFunction wrangler;
    };
 
    void CreateMudpiReceivers();
 
    void DeleteMudpiReceivers();
 
    void CreateDdsPubs();
    void DeleteDdsPubs();
 
    log4cplus::Logger& m_logger;
 
    const std::string m_name;
    uint16_t m_sim_freq;
 
    std::unique_ptr<DdsPub> m_dds_pub;
 
    rtctk::componentFramework::ServiceContainer& m_service_container;
 
    componentFramework::ComponentMetricsIf& m_component_metrics;
 
    std::vector<std::unique_ptr<UdpReceiver>> m_udp_receivers;
 
    std::vector<DdsPublisherContext> m_pub_threads;
 
    RepubWranglerFactory m_wrangler_factory;
};
 
}  // namespace rtctk::telRepub
 
#endif  // RTCTK_REUSABLECOMPONENT_TELREPUB_TELREPUBBUSINESSLOGIC_HPP