multiplex: Add functionality to rtcp_reader

include/uvgrtp/rtcp.hh

@@ -27,6 +27,7 @@ namespace uvgrtp {
     class srtcp;
     class socket;
     class socketfactory;
+    class rtcp_reader;
 
     typedef std::vector<std::pair<size_t, uint8_t*>> buf_vec; // also defined in socket.hh
 
@@ -653,6 +654,7 @@ namespace uvgrtp {
             std::unique_ptr<std::thread> report_reader_;
             std::shared_ptr<uvgrtp::socket> rtcp_socket_;
             std::shared_ptr<uvgrtp::socketfactory> sfp_;
+            std::shared_ptr<uvgrtp::rtcp_reader> rtcp_reader_;
 
             bool is_active() const
             {

src/rtcp.cc

@@ -11,6 +11,7 @@
 #include "srtp/srtcp.hh"
 #include "rtcp_packets.hh"
 #include "socketfactory.hh"
+#include "rtcp_reader.hh"
 
 #include "global.hh"
 
@@ -69,6 +70,7 @@ uvgrtp::rtcp::rtcp(std::shared_ptr<uvgrtp::rtp> rtp, std::shared_ptr<std::atomic
     app_hook_f_(nullptr),
     app_hook_u_(nullptr),
     sfp_(sfp),
+    rtcp_reader_(nullptr),
     active_(false),
     interval_ms_(DEFAULT_RTCP_INTERVAL_MS),
     rtp_ptr_(rtp),
@@ -209,14 +211,22 @@ rtp_error_t uvgrtp::rtcp::start()
     if (local_port_ != 0 && !sfp_->is_port_in_use(local_port_)) {
         rtcp_socket_ = sfp_->create_new_socket();
         new_socket_ = true;
+        rtcp_reader_ = std::shared_ptr<uvgrtp::rtcp_reader>(new uvgrtp::rtcp_reader(sfp_));
+
+        rtcp_reader_->map_ssrc_to_rtcp(ssrc_, std::shared_ptr<uvgrtp::rtcp>(this));
     }
     // Source port is in use -> fetch the existing socket
     else {
         rtcp_socket_ = sfp_->get_socket_ptr(local_port_);
+        // If socket still is null, create a new one
         if (!rtcp_socket_) {
             rtcp_socket_ = sfp_->create_new_socket();
             new_socket_ = true;
+            rtcp_reader_ = std::shared_ptr<uvgrtp::rtcp_reader>(new uvgrtp::rtcp_reader(sfp_));
+            rtcp_reader_->map_ssrc_to_rtcp(ssrc_, std::shared_ptr<uvgrtp::rtcp>(this));
         }
+        // Otherwise use the given existing socket
+
     }
 
     rtp_error_t ret = RTP_OK;

src/rtcp_reader.cc

@@ -25,13 +25,29 @@ uvgrtp::rtcp_reader::rtcp_reader(std::shared_ptr<uvgrtp::socketfactory> sfp) :
     socket_(nullptr),
     rtcps_map_({})
 {
-
+    report_reader_ = nullptr;
 }
 
 uvgrtp::rtcp_reader::~rtcp_reader()
 {
 }
 
+rtp_error_t uvgrtp::rtcp_reader::start()
+{
+    report_reader_.reset(new std::thread(rtcp_report_reader, this));
+
+}
+
+rtp_error_t uvgrtp::rtcp_reader::stop()
+{
+    if (report_reader_ && report_reader_->joinable())
+    {
+        UVG_LOG_DEBUG("Waiting for RTCP reader to exit");
+        report_reader_->join();
+    }
+
+}
+
 void uvgrtp::rtcp_reader::rtcp_report_reader() {
 
     UVG_LOG_INFO("RTCP report reader created!");
@@ -40,7 +56,6 @@ void uvgrtp::rtcp_reader::rtcp_report_reader() {
     rtp_error_t ret = RTP_OK;
     int max_poll_timeout_ms = 100;
 
-
     while (active_) {
         int nread = 0;
 
@@ -53,11 +68,11 @@ void uvgrtp::rtcp_reader::rtcp_report_reader() {
         {
             uint32_t sender_ssrc = ntohl(*(uint32_t*)&buffer.get()[0 + RTCP_HEADER_SIZE]);
             for (auto& p : rtcps_map_) {
-                if (sender_ssrc == p.first) {
+                if (sender_ssrc == p.first.get()->load()) {
                     std::shared_ptr<uvgrtp::rtcp> rtcp_ptr = p.second;
                     (void)rtcp_ptr->handle_incoming_packet(buffer.get(), (size_t)nread);
                 }
-           
+            
             }
         }
         else if (ret == RTP_INTERRUPTED) {
@@ -75,4 +90,10 @@ bool uvgrtp::rtcp_reader::set_socket(std::shared_ptr<uvgrtp::socket> socket)
 {
     socket_ = socket;
     return true;
+}
+
+bool uvgrtp::rtcp_reader::map_ssrc_to_rtcp(std::shared_ptr<std::atomic<uint32_t>> ssrc, std::shared_ptr<uvgrtp::rtcp> rtcp)
+{
+    rtcps_map_[ssrc] = rtcp;
+    return true;
 }

src/rtcp_reader.hh

@@ -14,6 +14,8 @@
 #include <vector>
 #include <memory>
 #include <map>
+#include <functional>
+
 
 namespace uvgrtp {
     class socketfactory;
@@ -25,14 +27,20 @@ namespace uvgrtp {
         public: 
             rtcp_reader(std::shared_ptr<uvgrtp::socketfactory> sfp);
             ~rtcp_reader();
+            rtp_error_t start();
+            rtp_error_t stop();
+
             void rtcp_report_reader();
             bool set_socket(std::shared_ptr<uvgrtp::socket> socket);
+            bool map_ssrc_to_rtcp(std::shared_ptr<std::atomic<uint32_t>> ssrc, std::shared_ptr<uvgrtp::rtcp> rtcp);
 
         private:
             bool active_;
             std::shared_ptr<uvgrtp::socketfactory> sfp_;
             std::shared_ptr<uvgrtp::socket> socket_;
-            std::map<uint32_t, std::shared_ptr<uvgrtp::rtcp>> rtcps_map_;
+            std::map<std::shared_ptr<std::atomic<uint32_t>>, std::shared_ptr<uvgrtp::rtcp>> rtcps_map_;
+            std::unique_ptr<std::thread> report_reader_;
+
     };