add kbps statistic for webrtc play

trunk/src/app/srs_app_listener.cpp

@@ -293,6 +293,8 @@ SrsUdpMuxSocket::SrsUdpMuxSocket(srs_netfd_t fd)
 
     fromlen = 0;
     peer_port = 0;
+
+    in_bytes = out_bytes = 0;
 }
 
 SrsUdpMuxSocket::~SrsUdpMuxSocket()
@@ -327,6 +329,7 @@ int SrsUdpMuxSocket::recvfrom(srs_utime_t timeout)
         peer_ip = std::string(address_string);
         peer_port = atoi(port_string);    
     }
+    in_bytes += nread;
 
     return nread;
 }
@@ -343,7 +346,8 @@ srs_error_t SrsUdpMuxSocket::sendto(void* data, int size, srs_utime_t timeout)
         }   
 
         return srs_error_new(ERROR_SOCKET_WRITE, "sendto");
-    }   
+    }  
+    out_bytes += nb_write;
 
     return err;
 }
@@ -408,6 +412,17 @@ SrsUdpMuxSocket* SrsUdpMuxSocket::copy_sendonly()
     return sendonly;
 }
 
+int64_t SrsUdpMuxSocket::get_recv_bytes()
+{
+    return in_bytes;
+}
+
+int64_t SrsUdpMuxSocket::get_send_bytes()
+{
+    return out_bytes;
+}
+
+
 SrsUdpMuxListener::SrsUdpMuxListener(ISrsUdpMuxHandler* h, std::string i, int p)
 {
     handler = h;

trunk/src/app/srs_app_listener.hpp

@@ -32,6 +32,7 @@
 #include <string>
 
 #include <srs_app_st.hpp>
+#include <srs_protocol_io.hpp>
 
 struct sockaddr;
 
@@ -133,8 +134,11 @@ class SrsTcpListener : public ISrsCoroutineHandler
 };
 
 // TODO: FIXME: Rename it. Refine it for performance issue.
-class SrsUdpMuxSocket
+class SrsUdpMuxSocket : public ISrsProtocolStatistic
 {
+private:
+    int64_t in_bytes;
+    int64_t out_bytes;
 private:
     char* buf;
     int nb_buf;
@@ -159,6 +163,9 @@ class SrsUdpMuxSocket
     int get_peer_port() const;
     std::string peer_id();
     SrsUdpMuxSocket* copy_sendonly();
+public:
+    virtual int64_t get_recv_bytes();
+    virtual int64_t get_send_bytes();
 };
 
 class SrsUdpMuxListener : public ISrsCoroutineHandler

trunk/src/app/srs_app_rtc_conn.cpp

@@ -616,6 +616,9 @@ srs_error_t SrsRtcPlayStream::cycle()
                 srs_freep(pkt);
             }
             pkts.clear();
+
+            // add webrtc kbps statistic
+            stat->kbps_add_delta(cid_, session_);
         }
 
         // Stat for performance analysis.
@@ -1613,6 +1616,9 @@ SrsRtcConnection::SrsRtcConnection(SrsRtcServer* s, const SrsContextId& cid)
     pli_epp = new SrsErrorPithyPrint();
 
     _srs_rtc_manager->subscribe(this);
+
+    clk = new SrsWallClock();
+    kbps = new SrsKbps(clk);
 }
 
 SrsRtcConnection::~SrsRtcConnection()
@@ -1650,6 +1656,13 @@ SrsRtcConnection::~SrsRtcConnection()
     srs_freep(stat_);
     srs_freep(pp_address_change);
     srs_freep(pli_epp);
+    srs_freep(kbps);
+    srs_freep(clk);
+}
+
+void SrsRtcConnection::remark(int64_t* in, int64_t* out)
+{
+    kbps->remark(in, out);
 }
 
 void SrsRtcConnection::on_before_dispose(ISrsResource* c)
@@ -2254,6 +2267,8 @@ void SrsRtcConnection::update_sendonly_socket(SrsUdpMuxSocket* skt)
 
     // Update the transport.
     sendonly_skt = addr_cache;
+
+    kbps->set_io(sendonly_skt, sendonly_skt);
 }
 
 srs_error_t SrsRtcConnection::notify(int type, srs_utime_t interval, srs_utime_t tick)

trunk/src/app/srs_app_rtc_conn.hpp

@@ -40,6 +40,7 @@
 #include <srs_app_rtc_dtls.hpp>
 #include <srs_service_conn.hpp>
 #include <srs_app_conn.hpp>
+#include <srs_protocol_kbps.hpp>
 
 #include <string>
 #include <map>
@@ -62,6 +63,7 @@ class SrsRtcConsumer;
 class SrsRtcAudioSendTrack;
 class SrsRtcVideoSendTrack;
 class SrsErrorPithyPrint;
+class SrsWallClock;
 
 const uint8_t kSR   = 200;
 const uint8_t kRR   = 201;
@@ -216,28 +218,28 @@ class SrsRtcPlayStreamStatistic
 {
 public:
     // The total bytes of AVFrame packets.
-    int nn_bytes;
+    uint64_t nn_bytes;
     // The total bytes of RTP packets.
-    int nn_rtp_bytes;
+    uint64_t nn_rtp_bytes;
     // The total padded bytes.
-    int nn_padding_bytes;
+    uint64_t nn_padding_bytes;
 public:
     // The RTP packets send out by sendmmsg or sendmsg. Note that if many packets group to
     // one msghdr by GSO, it's only one RTP packet, because we only send once.
-    int nn_rtp_pkts;
+    uint64_t nn_rtp_pkts;
     // For video, the samples or NALUs.
     // TODO: FIXME: Remove it because we may don't know.
-    int nn_samples;
+    uint64_t nn_samples;
     // For audio, the generated extra audio packets.
     // For example, when transcoding AAC to opus, may many extra payloads for a audio.
     // TODO: FIXME: Remove it because we may don't know.
-    int nn_extras;
+    uint64_t nn_extras;
     // The original audio messages.
-    int nn_audios;
+    uint64_t nn_audios;
     // The original video messages.
-    int nn_videos;
+    uint64_t nn_videos;
     // The number of padded packet.
-    int nn_paddings;
+    uint64_t nn_paddings;
 public:
     SrsRtcPlayStreamStatistic();
     virtual ~SrsRtcPlayStreamStatistic();
@@ -415,7 +417,7 @@ class ISrsRtcConnectionHijacker
 
 // A RTC Peer Connection, SDP level object.
 class SrsRtcConnection : virtual public ISrsHourGlass, virtual public ISrsResource
-    , virtual public ISrsDisposingHandler
+    , virtual public ISrsDisposingHandler, virtual public ISrsKbpsDelta
 {
     friend class SrsSecurityTransport;
     friend class SrsRtcPlayStream;
@@ -439,6 +441,12 @@ class SrsRtcConnection : virtual public ISrsHourGlass, virtual public ISrsResour
     // key: publisher track's ssrc
     std::map<uint32_t, SrsRtcPublishStream*> publishers_ssrc_map_;
 private:
+    // The connection total kbps.
+    // not only the rtmp or http connection, all type of connection are
+    // need to statistic the kbps of io.
+    // The SrsStatistic will use it indirectly to statistic the bytes delta of current connection.
+    SrsKbps* kbps;
+    SrsWallClock* clk;
     // The local:remote username, such as m5x0n128:jvOm where local name is m5x0n128.
     std::string username_;
     // The peer address, client maybe use more than one address, it's the current selected one.
@@ -470,6 +478,9 @@ class SrsRtcConnection : virtual public ISrsHourGlass, virtual public ISrsResour
 public:
     SrsRtcConnection(SrsRtcServer* s, const SrsContextId& cid);
     virtual ~SrsRtcConnection();
+// Interface ISrsKbpsDelta
+public:
+    virtual void remark(int64_t* in, int64_t* out);
 // interface ISrsDisposingHandler
 public:
     virtual void on_before_dispose(ISrsResource* c);