RTC: Stat the WebRTC clients bandwidth. v5.0.50

trunk/doc/CHANGELOG.md

@@ -7,6 +7,7 @@ The changelog for SRS.
 
 ## SRS 5.0 Changelog
 
+* v5.0, 2022-08-29, RTC: Stat the WebRTC clients bandwidth. v5.0.50
 * v5.0, 2022-08-29, HLS: Stat the HLS streaming clients bandwidth. v5.0.49
 * v5.0, 2022-08-28, URL: Use SrsHttpUri to parse URL and query. v5.0.48
 * v5.0, 2022-08-28, Fix [#2881](https://github.com/ossrs/srs/issues/2881): HTTP: Support merging api to server. v5.0.47

trunk/src/app/srs_app_rtc_conn.cpp

@@ -46,6 +46,7 @@ using namespace std;
 #include <srs_app_log.hpp>
 #include <srs_app_http_hooks.hpp>
 #include <srs_protocol_kbps.hpp>
+#include <srs_kernel_kbps.hpp>
 
 SrsPps* _srs_pps_sstuns = NULL;
 SrsPps* _srs_pps_srtcps = NULL;
@@ -1823,6 +1824,10 @@ SrsRtcConnection::SrsRtcConnection(SrsRtcServer* s, const SrsContextId& cid)
     nack_enabled_ = false;
     timer_nack_ = new SrsRtcConnectionNackTimer(this);
 
+    clock_ = new SrsWallClock();
+    kbps_ = new SrsKbps(clock_);
+    kbps_->set_io(NULL, NULL);
+
     _srs_rtc_manager->subscribe(this);
 }
 
@@ -1867,6 +1872,9 @@ SrsRtcConnection::~SrsRtcConnection()
     srs_freep(req_);
     srs_freep(pp_address_change);
     srs_freep(pli_epp);
+
+    srs_freep(kbps_);
+    srs_freep(clock_);
 }
 
 void SrsRtcConnection::on_before_dispose(ISrsResource* c)
@@ -1942,6 +1950,11 @@ vector<SrsUdpMuxSocket*> SrsRtcConnection::peer_addresses()
     return addresses;
 }
 
+void SrsRtcConnection::remark(int64_t* in, int64_t* out)
+{
+    kbps_->remark(in, out);
+}
+
 const SrsContextId& SrsRtcConnection::get_id()
 {
     return cid_;
@@ -2096,6 +2109,9 @@ srs_error_t SrsRtcConnection::on_stun(SrsUdpMuxSocket* skt, SrsStunPacket* r)
 {
     srs_error_t err = srs_success;
 
+    // Update stat when we received data.
+    kbps_->add_delta(skt->size(), 0);
+
     if (!r->is_binding_request()) {
         return err;
     }
@@ -2118,13 +2134,19 @@ srs_error_t SrsRtcConnection::on_stun(SrsUdpMuxSocket* skt, SrsStunPacket* r)
 
 srs_error_t SrsRtcConnection::on_dtls(char* data, int nb_data)
 {
+    // Update stat when we received data.
+    kbps_->add_delta(nb_data, 0);
+
     return transport_->on_dtls(data, nb_data);
 }
 
 srs_error_t SrsRtcConnection::on_rtcp(char* data, int nb_data)
 {
     srs_error_t err = srs_success;
 
+    // Update stat when we received data.
+    kbps_->add_delta(nb_data, 0);
+
     int nb_unprotected_buf = nb_data;
     if ((err = transport_->unprotect_rtcp(data, &nb_unprotected_buf)) != srs_success) {
         return srs_error_wrap(err, "rtcp unprotect");
@@ -2263,6 +2285,9 @@ srs_error_t SrsRtcConnection::on_rtp(char* data, int nb_data)
 {
     srs_error_t err = srs_success;
 
+    // Update stat when we received data.
+    kbps_->add_delta(nb_data, 0);
+
     SrsRtcPublishStream* publisher = NULL;
     if ((err = find_publisher(data, nb_data, &publisher)) != srs_success) {
         return srs_error_wrap(err, "find");
@@ -2459,6 +2484,9 @@ srs_error_t SrsRtcConnection::send_rtcp(char *data, int nb_data)
 
     ++_srs_pps_srtcps->sugar;
 
+    // Update stat when we sending data.
+    kbps_->add_delta(0, nb_data);
+
     int  nb_buf = nb_data;
     if ((err = transport_->protect_rtcp(data, &nb_buf)) != srs_success) {
         return srs_error_wrap(err, "protect rtcp");
@@ -2663,6 +2691,9 @@ srs_error_t SrsRtcConnection::do_send_packet(SrsRtpPacket* pkt)
 
     ++_srs_pps_srtps->sugar;
 
+    // Update stat when we sending data.
+    kbps_->add_delta(0, iov->iov_len);
+
     // TODO: FIXME: Handle error.
     sendonly_skt->sendto(iov->iov_base, iov->iov_len, 0);
 
@@ -2734,6 +2765,9 @@ srs_error_t SrsRtcConnection::on_binding_request(SrsStunPacket* r)
         return srs_error_wrap(err, "stun binding response encode failed");
     }
 
+    // Update stat when we sending data.
+    kbps_->add_delta(0, stream->pos());
+
     if ((err = sendonly_skt->sendto(stream->data(), stream->pos(), 0)) != srs_success) {
         return srs_error_wrap(err, "stun binding response send failed");
     }

trunk/src/app/srs_app_rtc_conn.hpp

@@ -51,6 +51,8 @@ class SrsStatistic;
 class SrsRtcUserConfig;
 class SrsRtcSendTrack;
 class SrsRtcPublishStream;
+class SrsKbps;
+class SrsWallClock;
 
 const uint8_t kSR   = 200;
 const uint8_t kRR   = 201;
@@ -433,7 +435,7 @@ class SrsRtcConnectionNackTimer : public ISrsFastTimer
 //
 // For performance, we use non-public from resource,
 // see https://stackoverflow.com/questions/3747066/c-cannot-convert-from-base-a-to-derived-type-b-via-virtual-base-a
-class SrsRtcConnection : public ISrsResource, public ISrsDisposingHandler, public ISrsExpire
+class SrsRtcConnection : public ISrsResource, public ISrsDisposingHandler, public ISrsExpire, public ISrsKbpsDelta
 {
     friend class SrsSecurityTransport;
     friend class SrsRtcPlayStream;
@@ -490,6 +492,9 @@ class SrsRtcConnection : public ISrsResource, public ISrsDisposingHandler, publi
     SrsErrorPithyPrint* pli_epp;
 private:
     bool nack_enabled_;
+private:
+    SrsKbps* kbps_;
+    SrsWallClock* clock_;
 public:
     SrsRtcConnection(SrsRtcServer* s, const SrsContextId& cid);
     virtual ~SrsRtcConnection();
@@ -510,6 +515,9 @@ class SrsRtcConnection : public ISrsResource, public ISrsDisposingHandler, publi
     std::string username();
     // Get all addresses client used.
     std::vector<SrsUdpMuxSocket*> peer_addresses();
+// Interface ISrsKbpsDelta.
+public:
+    virtual void remark(int64_t* in, int64_t* out);
 // Interface ISrsResource.
 public:
     virtual const SrsContextId& get_id();

trunk/src/app/srs_app_rtc_server.cpp

@@ -643,6 +643,10 @@ srs_error_t SrsRtcServer::on_timer(srs_utime_t interval)
         // Update stat if session is alive.
         if (session->is_alive()) {
             nn_rtc_conns++;
+
+            ISrsKbpsDelta* conn = dynamic_cast<ISrsKbpsDelta*>(session);
+            SrsStatistic::instance()->kbps_add_delta(session->get_id().c_str(), conn);
+
             continue;
         }
 

trunk/src/core/srs_core_version5.hpp

@@ -9,6 +9,6 @@
 
 #define VERSION_MAJOR       5
 #define VERSION_MINOR       0
-#define VERSION_REVISION    49
+#define VERSION_REVISION    50
 
 #endif