rtmsReceiver.hpp

Go to the documentation of this file.

 
#pragma once
 
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <csignal>
#include <cstddef>
#include <mutex>
#include <string>
#include <thread>
#include <vector>
 
#include <boost/endian/arithmetic.hpp>
 
#include <fmt/format.h>
 
#include <llnetio/ipv4.hpp>
#include <llnetio/mudpi/io.hpp>
#include <llnetio/rtms/io.hpp>
#include <llnetio/rtms/rtms.hpp>
#include <llnetio/udpSocket.hpp>
#include <llnetio/mockSocket.hpp>
 
#include <ifw/rtmslib_llnetio/logger.hpp>
#include <ifw/rtmslib_llnetio/common.hpp>
#include <ifw/rtmslib_llnetio/sampleExtInfo.hpp>
 
namespace ifw::rtmslib {
 
    enum class SampleFailureKind {
        None,
        Timeout,
        SampleError
    };
 
    struct ReceiverConfig {
        std::size_t      rxbuf_size{16777216};
        int32_t          topic_id{10};
        uint16_t         mtu_size{1500};
        int32_t          width{0};
        int32_t          height{0};
        bool             checksum_force_ignore{false};
        ExtendedInfoMode ext_info_mode{ExtendedInfoMode::None};
    };

    class RtmsReceiver {
      public:
        static std::atomic<bool>    s_execute;

        RtmsReceiver(const std::string& receiver_ip,
                     const std::string& interface_name,                     
                     const int port,
                     const ifw::fnd::datatype::DataType data_type,
                     const ReceiverConfig& rcv_cfg = ReceiverConfig());
 
        virtual ~RtmsReceiver();

        void ReceiveLoop();

        virtual void StopReceiver();

        bool ReceiveSample(SampleExtInfo& ext_info,
                           std::vector<uint8_t>& sample_payload,
                           std::chrono::nanoseconds timeout);

        virtual void HandleSampleUser(SampleExtInfo& ext_info,
                                      std::vector<uint8_t>& sample_payload) = 0;
 
        int64_t GetLostFramesCount() const;
 
    protected:
 
      int64_t GetSampleId() const;
      ifw::fnd::datatype::DataType GetDataType() const;
      ReceiverConfig GetConfig() const;
      void RequestStop();
 
    private:
      void EnsureInitialised();
      void Initialise();
 
      std::string                               m_receiver_ip;
      std::string                               m_interface_name;
      int                                       m_port;
      ifw::fnd::datatype::DataType              m_data_type;
      int8_t                                    m_bpp;
      std::once_flag                            m_init_once;
 
      ReceiverConfig                            m_rcv_cfg;
      uint16_t                                  m_leader_extinfo_capacity{
                                                  static_cast<uint16_t>(SIZE_SAMPLE_EXT_INFO)};
      ExtendedInfoMode                          m_current_extinfo_mode{ExtendedInfoMode::None};
      SampleFailureKind                         m_last_failure_kind{SampleFailureKind::None};
      int64_t                                   m_samples_received{0};
      int64_t                                   m_sample_failures{0};
      int64_t                                   m_sample_timeouts{0};
      std::chrono::steady_clock::time_point     m_last_status_log{};
 
      std::unique_ptr<llnetio::rtms::Receiver>  m_rtms_receiver;
      std::string                               m_last_receive_error;
      std::chrono::steady_clock::time_point     m_last_loss_log{};
      std::chrono::steady_clock::time_point     m_last_sample_log{};
      std::chrono::steady_clock::time_point     m_last_extinfo_log{};
 
      int64_t                                   m_lost_frames{0};
      unsigned int                              m_sample_id{0};
      int64_t                                   m_extinfo_mismatches{0};
      int                                       m_consecutive_leader_timeouts{0};
 
      llnetio::UdpRxSocket* m_udp_sock;
      std::unique_ptr<llnetio::mudpi::Receiver> m_mudpi;
 
      std::atomic<bool>                         m_ongoing_recording;
      std::mutex                                m_recording_mutex;
      std::condition_variable                   m_recording_cv;
      bool                                      m_dynamic_sample_bytes_logged{false};
      bool                                      m_missing_bpp_logged{false};
 
      void ResetTransport();
      void LogStatus(bool force = false);
      bool UpdateGeometryFromExtInfo(const SampleExtInfo& ext_info);
      std::size_t ComputeSampleBytesHint() const;
 
    };
}
 
inline void IfwRtmsLibStopExecution(int signum) {
  ifw::rtmslib::RtmsReceiver::s_execute = false;
}