Merge pull request #10 from Unit-X/star-486-add-statistics-callback

STAR-486 Implement support for statistic callbacks

Author: tomasoh-ateliere

RISTNet.cpp

@@ -173,6 +173,14 @@ int RISTNetReceiver::clientDisconnect(void *pArg, rist_peer *pPeer) {
     return 0;
 }
 
+int RISTNetReceiver::gotStatistics(void *pArg, const rist_stats *stats) {
+    RISTNetReceiver *lWeakSelf = static_cast<RISTNetReceiver*>(pArg);
+    if (lWeakSelf->statisticsCallback) {
+        lWeakSelf->statisticsCallback(*stats);
+    }
+    return rist_stats_free(stats);
+}
+
 //---------------------------------------------------------------------------------------------------------------------
 // RISTNetReceiver  --  Callbacks --- End
 //---------------------------------------------------------------------------------------------------------------------
@@ -333,6 +341,13 @@ bool RISTNetReceiver::initReceiver(std::vector<std::string> &rURLList,
         return false;
     }
 
+    lStatus = rist_stats_callback_set(mRistContext, 1000, gotStatistics, this);
+    if (lStatus) {
+        LOGGER(true, LOGG_ERROR, "rist_stats_callback_set fail.")
+        destroyReceiver();
+        return false;
+    }
+
     lStatus = rist_start(mRistContext);
     if (lStatus) {
         LOGGER(true, LOGG_ERROR, "rist_receiver_start fail.")
@@ -456,6 +471,14 @@ int RISTNetSender::clientDisconnect(void *pArg, rist_peer *pPeer) {
     return 0;
 }
 
+int RISTNetSender::gotStatistics(void *pArg, const rist_stats *stats) {
+    RISTNetSender *lWeakSelf = static_cast<RISTNetSender*>(pArg);
+    if (lWeakSelf->statisticsCallback) {
+        lWeakSelf->statisticsCallback(*stats);
+    }
+    return rist_stats_free(stats);
+}
+
 //---------------------------------------------------------------------------------------------------------------------
 // RISTNetSender  --  Callbacks --- End
 //---------------------------------------------------------------------------------------------------------------------
@@ -609,6 +632,13 @@ bool RISTNetSender::initSender(std::vector<std::tuple<std::string,int>> &rPeerLi
         return false;
     }
 
+    lStatus = rist_stats_callback_set(mRistContext, 1000, gotStatistics, this);
+    if (lStatus) {
+        LOGGER(true, LOGG_ERROR, "rist_stats_callback_set fail.")
+        destroySender();
+        return false;
+    }
+
     lStatus = rist_start(mRistContext);
     if (lStatus) {
         LOGGER(true, LOGG_ERROR, "rist_sender_start fail.")

RISTNet.h

@@ -196,9 +196,6 @@ class RISTNetReceiver {
    */
   static void getVersion(uint32_t &rCppWrapper, uint32_t &rRistMajor, uint32_t &rRistMinor);
 
-  //To be implemented
-  //void getInfo();
-
   /**
    * @brief Data receive callback
    *
@@ -242,6 +239,9 @@ class RISTNetReceiver {
   /// Callback handling disconnecting clients
   std::function<void(const std::shared_ptr<NetworkConnection>&, const rist_peer&)> clientDisconnectedCallback = nullptr;
 
+  /// Callback for statistics, called once every second
+  std::function<void(const rist_stats& statistics)> statisticsCallback = nullptr;
+
   // Delete copy and move constructors and assign operators
   RISTNetReceiver(RISTNetReceiver const &) = delete;             // Copy construct
   RISTNetReceiver(RISTNetReceiver &&) = delete;                  // Move construct
@@ -265,11 +265,14 @@ class RISTNetReceiver {
   // Private method called when a client disconnects
   static int clientDisconnect(void *pArg, rist_peer *pPeer);
 
+  // Private method called when a statistics are delivered
+  static int gotStatistics(void *pArg, const rist_stats *stats);
+
   // The context of a RIST receiver
   rist_ctx *mRistContext = nullptr;
 
   // The configuration of the RIST receiver
-  rist_peer_config mRistPeerConfig = {0};
+  rist_peer_config mRistPeerConfig{};
 
   // The mutex protecting the list. since the list can be accessed from both librist and the C++ layer
   std::mutex mClientListMtx;
@@ -434,9 +437,6 @@ class RISTNetSender {
    */
    static void getVersion(uint32_t &rCppWrapper, uint32_t &rRistMajor, uint32_t &rRistMinor);
 
-  //To be implemented
-  //void getInfo();
-
   /**
    * @brief OOB Data receive callback (__NULLABLE)
    *
@@ -467,6 +467,9 @@ class RISTNetSender {
   /// Callback handling disconnecting clients
   std::function<void(const std::shared_ptr<NetworkConnection>&, const rist_peer&)> clientDisconnectedCallback = nullptr;
 
+  /// Callback for statistics, called once every second
+  std::function<void(const rist_stats& statistics)> statisticsCallback = nullptr;
+
   // Delete copy and move constructors and assign operators
   RISTNetSender(RISTNetSender const &) = delete;             // Copy construct
   RISTNetSender(RISTNetSender &&) = delete;                  // Move construct
@@ -487,11 +490,14 @@ class RISTNetSender {
   // Private method called when a client disconnects
   static int clientDisconnect(void *pArg, rist_peer *pPeer);
 
+  // Private method called when statistics are delivered
+  static int gotStatistics(void *pArg, const rist_stats *stats);
+
   // The context of a RIST sender
   rist_ctx *mRistContext = nullptr;
 
   // The configuration of the RIST sender
-  rist_peer_config mRistPeerConfig = {0};
+  rist_peer_config mRistPeerConfig{};
 
   // The mutex protecting the list. since the list can be accessed from both librist and the C++ layer
   std::mutex mClientListMtx;

test/TestRist.cpp

@@ -213,13 +213,29 @@ TEST_F(TestFixture, SendReceive) {
         return 0;
     };
 
+    std::atomic<size_t> receiverStatisticCounter = 0;
+    mReceiver->statisticsCallback = [&](const rist_stats& stats) {
+        receiverStatisticCounter++;
+    };
+    std::atomic<size_t> senderStatisticCounter = 0;
+    mSender->statisticsCallback = [&](const rist_stats& stats) {
+        senderStatisticCounter++;
+    };
+
     std::vector<uint8_t> sendBuffer(kBufferSize);
     for (auto i = 0; i < kSentPackets; i++) {
         std::this_thread::sleep_for(std::chrono::milliseconds(50));
         std::fill(sendBuffer.begin(), sendBuffer.end(), '0' + i);
         EXPECT_TRUE(mSender->sendData(static_cast<const uint8_t*>(sendBuffer.data()), sendBuffer.size()));
     }
 
+    std::this_thread::sleep_for(std::chrono::seconds(10));
+
+    size_t numCallsReceiver = receiverStatisticCounter;
+    EXPECT_GE(numCallsReceiver, 9) << "Expected the statistics to be delivered at least 9 times for receiver";
+    size_t numCallsSender = senderStatisticCounter;
+    EXPECT_GE(numCallsSender, 9) << "Expected the statistics to be delivered at least 9 times for sender";
+
     {
         std::unique_lock<std::mutex> lock(receiverMutex);
         bool successfulWait =