rtmsPublisher.hpp

Go to the documentation of this file.

 
#ifndef FGF_COM_RTMS_PUBLISHER_HPP_H_
#define FGF_COM_RTMS_PUBLISHER_HPP_H_
 
#define BOOST_BIND_NO_PLACEHOLDERS
 
#include <list>
 
#include <boost/lockfree/spsc_queue.hpp>
#include <boost/atomic.hpp>
 
#include <mudpiif/MudpiSender.hpp>
 
#include <ifw/fgf/common/logger.hpp>
#include <ifw/fgf/common/defines.hpp>
#include <ifw/fgf/common/rawImage.hpp>
#include <ifw/fgf/common/binarySemaphore.hpp>
#include <ifw/fgf/common/mudpiCreatorParallel.hpp>
 
namespace ifw::fgf::common {
 
    constexpr auto STAT_WINDOW_SIZE = 1000;
 
    class RtmsPublisher {
    public:
 
        RtmsPublisher(const std::string address,
                      const int port,
                      const std::string netif = "",
                      const double packet_send_delay = 0);
 
        virtual ~RtmsPublisher();
 
        void StartPubThread();
        void StopPubThread();
        void StartPublishing();
        void InitPublisher(size_t raw_image_size, int bytes_per_pixel);
 
        void Run();
 
        void ComputeLatency();
        bool GetComputeLatency() const;
        void ResetLatency();
        void GetFirstPacketSentLatency(double& latency, double& stddev, double& variance,
                                      int& nb_of_samples);
        void GetFrameSentLatency(double& latency, double& stddev, double& variance,
                                 int& nb_of_samples);
 
    public:
        boost::asio::io_context                    m_ioc;
        std::string                                m_address{""};
        int                                        m_port{0};
        std::string                                m_netif{""};
 
        // Basic MUDPI configuration
        int                                        m_topicid{10};
        int                                        m_sampleid{500};
        std::unique_ptr<ifw::fgf::common::MudpiCreatorParallel> m_mudpi_creator;
        std::unique_ptr<mudpiif::mudpiSenderJumbo> m_mudpi_sender;
        static std::atomic<bool>                   m_stopped;  // Flag to stop infinite loop
 
        // Single-writer/single-reader FIFO queue.
        static boost::lockfree::spsc_queue<std::shared_ptr<RawImage>,
            boost::lockfree::capacity<5>> spsc_queue;
 
    private:
        double                         m_packet_send_delay{0};
        std::chrono::duration<double>  m_chrono_sleep_time;
 
        std::thread                    m_pub_thread;
        std::condition_variable        m_pub_condvar;
        std::mutex                     m_mtx;
        log4cplus::Logger              m_logger;
 
        // Used to compute the latency.
        bool              m_compute_latency{false};
        std::mutex        m_start_pub_times_mtx;
        std::list<double> m_start_pub_times;
 
        std::mutex        m_first_packet_sent_latencies_mtx;
        std::list<double> m_first_packet_sent_latencies;
 
        std::mutex        m_frame_sent_latencies_mtx;
        std::list<double> m_frame_sent_latencies;
 
        void ComputeStat(const std::list<double>& samples,
                         double& mean,
                         double& stddev,
                         double& variance,
                         int& nb_of_samples) const;
 
    };
}
 
#endif  // FGF_COM_RTMS_PUBLISHER_HPP_H_