udpReceiver.hpp

Go to the documentation of this file.

#ifndef RTCTK_REUSABLECOMPONENT_TELREPUB_UDPRECEIVER_HPP
#define RTCTK_REUSABLECOMPONENT_TELREPUB_UDPRECEIVER_HPP
 
#include <boost/asio/io_service.hpp>
#include <chrono>
#include <cstdint>
#include <memory>
#include <thread>
 
#include <rtctk/componentFramework/alertServiceIf.hpp>
#include <rtctk/componentFramework/serviceContainer.hpp>
 
#include <boost/asio/io_context.hpp>
#include <boost/asio/ip/udp.hpp>
#include <log4cplus/logger.h>
 
#include "perfc/counter.hpp"
#include "queue.hpp"
#include "rtctk/componentFramework/durationMonitor.hpp"
#include <numapp/numapolicies.hpp>
 
namespace rtctk::telRepub {

struct UdpReceiverCfg {
    boost::asio::ip::udp::endpoint endpoint;
    boost::asio::ip::udp::endpoint multicast_group;
    numapp::NumaPolicies thread_policies;
    int buffer_size_bytes = 0;
    std::chrono::milliseconds receive_timeout_ms = std::chrono::milliseconds(20000);  // default 20s
};

constexpr uint16_t DATAGRAM_SIZE = 9500;

class UdpReceiver {
public:
    explicit UdpReceiver(const std::string& receiver_name,
                         std::map<int32_t, QueuePtr> queue_map,
                         size_t buffer_count,
                         componentFramework::ServiceContainer& service,
                         const UdpReceiverCfg& cfg);
    ~UdpReceiver() noexcept;

    void Enable();

    void Disable();

    bool CheckReceiveTimeout();
 
private:
    enum class State : uint8_t { Idle, Run, Exit };
    void InitReceive();
    void HandleReceive(const boost::system::error_code& error, std::size_t bytes_transferred);
 
    log4cplus::Logger& m_logger;

    boost::asio::io_service m_io_context;
    boost::asio::ip::udp::socket m_socket;
    boost::asio::ip::udp::endpoint m_sender;
    std::thread m_rcv_thr;
 
    std::atomic<State> m_requested_state;

    static std::atomic<std::uint16_t> s_count;
    std::string m_receiver_name;
    std::map<int32_t, QueuePtr> m_queue_map;
 
    componentFramework::ComponentMetricsIf& m_metrics;
    componentFramework::AlertServiceIf& m_alerts;
 
    perfc::CounterU64 m_frame_counter;
    perfc::ScopedRegistration m_frame_counter_registration;
 
    perfc::CounterU64 m_frames_dropped;
    perfc::ScopedRegistration m_frames_dropped_registration;
    std::atomic<std::chrono::steady_clock::time_point> m_last_receive;
 
    std::unique_ptr<componentFramework::DurationMonitor<>> m_frame_last_diff_monitor;
    std::unique_ptr<componentFramework::DurationMonitor<>> m_frame_receive_time_monitor;
 
    componentFramework::BufferManager m_buffer_manager;
    std::shared_ptr<componentFramework::BufferManager::BufferReference> m_current_buffer;
 
    // receive timeout alert
    componentFramework::AlertSource m_receive_timeout_alert;
 
    std::chrono::milliseconds m_receive_timeout_ms;
};
 
}  // namespace rtctk::telRepub
 
#endif  // RTCTK_REUSABLECOMPONENT_TELREPUB_UDPRECEIVER_HPP