refine framework to calc the kbps

@@ -464,11 +464,11 @@ MODULE_ID="APP"
 MODULE_DEPENDS=("CORE" "KERNEL" "RTMP") 
 MODULE_DEPENDS=("CORE" "KERNEL" "RTMP") 
 ModuleLibIncs=(${LibSTRoot} ${LibHttpParserRoot} ${SRS_OBJS})
 ModuleLibIncs=(${LibSTRoot} ${LibHttpParserRoot} ${SRS_OBJS})
 MODULE_FILES=("srs_app_server" "srs_app_conn" "srs_app_rtmp_conn" "srs_app_socket" "srs_app_source" 
 MODULE_FILES=("srs_app_server" "srs_app_conn" "srs_app_rtmp_conn" "srs_app_socket" "srs_app_source" 
-        "srs_app_codec" "srs_app_refer" "srs_app_hls" "srs_app_forward" "srs_app_encoder"
-        "srs_app_http" "srs_app_thread" "srs_app_bandwidth" "srs_app_st" "srs_app_log"
-        "srs_app_config" "srs_app_pithy_print" "srs_app_reload" "srs_app_http_api"
-        "srs_app_http_conn" "srs_app_http_hooks" "srs_app_json" "srs_app_ingest"
-        "srs_app_ffmpeg" "srs_app_utility" "srs_app_dvr" "srs_app_edge")
+        "srs_app_codec" "srs_app_refer" "srs_app_hls" "srs_app_forward" "srs_app_encoder" "srs_app_http" 
+        "srs_app_thread" "srs_app_bandwidth" "srs_app_st" "srs_app_log" "srs_app_config" 
+        "srs_app_pithy_print" "srs_app_reload" "srs_app_http_api" "srs_app_http_conn" "srs_app_http_hooks" 
+        "srs_app_json" "srs_app_ingest" "srs_app_ffmpeg" "srs_app_utility" "srs_app_dvr" "srs_app_edge"
+        "srs_app_kbps")
 APP_INCS="src/app"; MODULE_DIR=${APP_INCS} . auto/modules.sh
 APP_INCS="src/app"; MODULE_DIR=${APP_INCS} . auto/modules.sh
 APP_OBJS="${MODULE_OBJS[@]}"
 APP_OBJS="${MODULE_OBJS[@]}"
 #
 #

@@ -42,6 +42,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <srs_app_pithy_print.hpp>
 #include <srs_app_pithy_print.hpp>
 #include <srs_core_autofree.hpp>
 #include <srs_core_autofree.hpp>
 #include <srs_app_socket.hpp>
 #include <srs_app_socket.hpp>
+#include <srs_app_kbps.hpp>
 
 
 // when error, edge ingester sleep for a while and retry.
 // when error, edge ingester sleep for a while and retry.
 #define SRS_EDGE_INGESTER_SLEEP_US (int64_t)(1*1000*1000LL)
 #define SRS_EDGE_INGESTER_SLEEP_US (int64_t)(1*1000*1000LL)
@@ -61,6 +62,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 SrsEdgeIngester::SrsEdgeIngester()
 SrsEdgeIngester::SrsEdgeIngester()
 {
 {
     io = NULL;
     io = NULL;
+    kbps = new SrsKbps();
     client = NULL;
     client = NULL;
     _edge = NULL;
     _edge = NULL;
     _req = NULL;
     _req = NULL;
@@ -75,6 +77,7 @@ SrsEdgeIngester::~SrsEdgeIngester()
     stop();
     stop();
     
     
     srs_freep(pthread);
     srs_freep(pthread);
+    srs_freep(kbps);
 }
 }
 
 
 int SrsEdgeIngester::initialize(SrsSource* source, SrsPlayEdge* edge, SrsRequest* req)
 int SrsEdgeIngester::initialize(SrsSource* source, SrsPlayEdge* edge, SrsRequest* req)
@@ -101,6 +104,7 @@ void SrsEdgeIngester::stop()
     
     
     srs_freep(client);
     srs_freep(client);
     srs_freep(io);
     srs_freep(io);
+    kbps->set_io(NULL, NULL);
 }
 }
 
 
 int SrsEdgeIngester::cycle()
 int SrsEdgeIngester::cycle()
@@ -169,9 +173,8 @@ int SrsEdgeIngester::ingest()
         // pithy print
         // pithy print
         if (pithy_print.can_print()) {
         if (pithy_print.can_print()) {
             srs_trace("<- "SRS_LOG_ID_EDGE_PLAY
             srs_trace("<- "SRS_LOG_ID_EDGE_PLAY
-                " time=%"PRId64", obytes=%"PRId64", ibytes=%"PRId64", okbps=%d, ikbps=%d", 
-                pithy_print.age(), client->get_send_bytes(), client->get_recv_bytes(), 
-                client->get_send_kbps(), client->get_recv_kbps());
+                " time=%"PRId64", okbps=%d, ikbps=%d", 
+                pithy_print.age(), kbps->get_send_kbps(), kbps->get_recv_kbps());
         }
         }
 
 
         // read from client.
         // read from client.
@@ -303,6 +306,7 @@ int SrsEdgeIngester::connect_server()
     
     
     io = new SrsSocket(stfd);
     io = new SrsSocket(stfd);
     client = new SrsRtmpClient(io);
     client = new SrsRtmpClient(io);
+    kbps->set_io(io, io);
     
     
     // connect to server.
     // connect to server.
     std::string ip = srs_dns_resolve(server);
     std::string ip = srs_dns_resolve(server);
@@ -330,6 +334,7 @@ int SrsEdgeIngester::connect_server()
 SrsEdgeForwarder::SrsEdgeForwarder()
 SrsEdgeForwarder::SrsEdgeForwarder()
 {
 {
     io = NULL;
     io = NULL;
+    kbps = NULL;
     client = NULL;
     client = NULL;
     _edge = NULL;
     _edge = NULL;
     _req = NULL;
     _req = NULL;
@@ -347,6 +352,7 @@ SrsEdgeForwarder::~SrsEdgeForwarder()
     
     
     srs_freep(pthread);
     srs_freep(pthread);
     srs_freep(queue);
     srs_freep(queue);
+    srs_freep(kbps);
 }
 }
 
 
 void SrsEdgeForwarder::set_queue_size(double queue_size)
 void SrsEdgeForwarder::set_queue_size(double queue_size)
@@ -411,6 +417,7 @@ void SrsEdgeForwarder::stop()
     
     
     srs_freep(client);
     srs_freep(client);
     srs_freep(io);
     srs_freep(io);
+    kbps->set_io(NULL, NULL);
 }
 }
 
 
 int SrsEdgeForwarder::cycle()
 int SrsEdgeForwarder::cycle()
@@ -458,9 +465,8 @@ int SrsEdgeForwarder::cycle()
         // pithy print
         // pithy print
         if (pithy_print.can_print()) {
         if (pithy_print.can_print()) {
             srs_trace("-> "SRS_LOG_ID_EDGE_PUBLISH
             srs_trace("-> "SRS_LOG_ID_EDGE_PUBLISH
-                " time=%"PRId64", msgs=%d, obytes=%"PRId64", ibytes=%"PRId64", okbps=%d, ikbps=%d", 
-                pithy_print.age(), count, client->get_send_bytes(), client->get_recv_bytes(), 
-                client->get_send_kbps(), client->get_recv_kbps());
+                " time=%"PRId64", msgs=%d, okbps=%d, ikbps=%d", 
+                pithy_print.age(), count, kbps->get_send_kbps(), kbps->get_recv_kbps());
         }
         }
         
         
         // ignore when no messages.
         // ignore when no messages.
@@ -576,6 +582,7 @@ int SrsEdgeForwarder::connect_server()
     
     
     io = new SrsSocket(stfd);
     io = new SrsSocket(stfd);
     client = new SrsRtmpClient(io);
     client = new SrsRtmpClient(io);
+    kbps->set_io(io, io);
     
     
     // connect to server.
     // connect to server.
     std::string ip = srs_dns_resolve(server);
     std::string ip = srs_dns_resolve(server);

@@ -43,6 +43,7 @@ class SrsRtmpClient;
 class SrsMessage;
 class SrsMessage;
 class SrsMessageQueue;
 class SrsMessageQueue;
 class ISrsProtocolReaderWriter;
 class ISrsProtocolReaderWriter;
+class SrsKbps;
 
 
 /**
 /**
 * the state of edge, auto machine
 * the state of edge, auto machine
@@ -83,6 +84,7 @@ private:
     SrsThread* pthread;
     SrsThread* pthread;
     st_netfd_t stfd;
     st_netfd_t stfd;
     ISrsProtocolReaderWriter* io;
     ISrsProtocolReaderWriter* io;
+    SrsKbps* kbps;
     SrsRtmpClient* client;
     SrsRtmpClient* client;
     int origin_index;
     int origin_index;
 public:
 public:
@@ -116,6 +118,7 @@ private:
     SrsThread* pthread;
     SrsThread* pthread;
     st_netfd_t stfd;
     st_netfd_t stfd;
     ISrsProtocolReaderWriter* io;
     ISrsProtocolReaderWriter* io;
+    SrsKbps* kbps;
     SrsRtmpClient* client;
     SrsRtmpClient* client;
     int origin_index;
     int origin_index;
     /**
     /**

@@ -39,6 +39,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <srs_protocol_rtmp_stack.hpp>
 #include <srs_protocol_rtmp_stack.hpp>
 #include <srs_protocol_utility.hpp>
 #include <srs_protocol_utility.hpp>
 #include <srs_protocol_rtmp.hpp>
 #include <srs_protocol_rtmp.hpp>
+#include <srs_app_kbps.hpp>
 
 
 // when error, forwarder sleep for a while and retry.
 // when error, forwarder sleep for a while and retry.
 #define SRS_FORWARDER_SLEEP_US (int64_t)(3*1000*1000LL)
 #define SRS_FORWARDER_SLEEP_US (int64_t)(3*1000*1000LL)
@@ -50,6 +51,7 @@ SrsForwarder::SrsForwarder(SrsSource* _source)
     io = NULL;
     io = NULL;
     client = NULL;
     client = NULL;
     stfd = NULL;
     stfd = NULL;
+    kbps = new SrsKbps();
     stream_id = 0;
     stream_id = 0;
 
 
     pthread = new SrsThread(this, SRS_FORWARDER_SLEEP_US);
     pthread = new SrsThread(this, SRS_FORWARDER_SLEEP_US);
@@ -64,6 +66,7 @@ SrsForwarder::~SrsForwarder()
     srs_freep(pthread);
     srs_freep(pthread);
     srs_freep(queue);
     srs_freep(queue);
     srs_freep(jitter);
     srs_freep(jitter);
+    srs_freep(kbps);
 }
 }
 
 
 void SrsForwarder::set_queue_size(double queue_size)
 void SrsForwarder::set_queue_size(double queue_size)
@@ -146,6 +149,7 @@ void SrsForwarder::on_unpublish()
     
     
     srs_freep(client);
     srs_freep(client);
     srs_freep(io);
     srs_freep(io);
+    kbps->set_io(NULL, NULL);
 }
 }
 
 
 int SrsForwarder::on_meta_data(SrsSharedPtrMessage* metadata)
 int SrsForwarder::on_meta_data(SrsSharedPtrMessage* metadata)
@@ -275,6 +279,7 @@ int SrsForwarder::connect_server()
     
     
     io = new SrsSocket(stfd);
     io = new SrsSocket(stfd);
     client = new SrsRtmpClient(io);
     client = new SrsRtmpClient(io);
+    kbps->set_io(io, io);
     
     
     // connect to server.
     // connect to server.
     std::string ip = srs_dns_resolve(server);
     std::string ip = srs_dns_resolve(server);
@@ -338,9 +343,8 @@ int SrsForwarder::forward()
         // pithy print
         // pithy print
         if (pithy_print.can_print()) {
         if (pithy_print.can_print()) {
             srs_trace("-> "SRS_LOG_ID_FOWARDER
             srs_trace("-> "SRS_LOG_ID_FOWARDER
-                " time=%"PRId64", msgs=%d, obytes=%"PRId64", ibytes=%"PRId64", okbps=%d, ikbps=%d", 
-                pithy_print.age(), count, client->get_send_bytes(), client->get_recv_bytes(), 
-                client->get_send_kbps(), client->get_recv_kbps());
+                " time=%"PRId64", msgs=%d, okbps=%d, ikbps=%d", 
+                pithy_print.age(), count, kbps->get_send_kbps(), kbps->get_recv_kbps());
         }
         }
         
         
         // ignore when no messages.
         // ignore when no messages.

@@ -42,6 +42,7 @@ class SrsRtmpJitter;
 class SrsRtmpClient;
 class SrsRtmpClient;
 class SrsRequest;
 class SrsRequest;
 class SrsSource;
 class SrsSource;
+class SrsKbps;
 
 
 /**
 /**
 * forward the stream to other servers.
 * forward the stream to other servers.
@@ -61,6 +62,7 @@ private:
 private:
 private:
     SrsSource* source;
     SrsSource* source;
     ISrsProtocolReaderWriter* io;
     ISrsProtocolReaderWriter* io;
+    SrsKbps* kbps;
     SrsRtmpClient* client;
     SrsRtmpClient* client;
     SrsRtmpJitter* jitter;
     SrsRtmpJitter* jitter;
     SrsMessageQueue* queue;
     SrsMessageQueue* queue;

+/*
+The MIT License (MIT)
+
+Copyright (c) 2013-2014 winlin
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of
+this software and associated documentation files (the "Software"), to deal in
+the Software without restriction, including without limitation the rights to
+use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+the Software, and to permit persons to whom the Software is furnished to do so,
+subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+*/
+
+#include <srs_app_kbps.hpp>
+
+#include <srs_kernel_error.hpp>
+#include <srs_kernel_log.hpp>
+#include <srs_protocol_io.hpp>
+
+SrsKbps::SrsKbps()
+{
+    _in = NULL;
+    _out = NULL;
+}
+
+SrsKbps::~SrsKbps()
+{
+}
+
+void SrsKbps::set_io(ISrsProtocolReader* in, ISrsProtocolWriter* out)
+{
+    _in = in;
+    _out = out;
+}
+
+int SrsKbps::get_send_kbps()
+{
+    return 0;
+}
+
+int SrsKbps::get_recv_kbps()
+{
+    return 0;
+}
+

+/*
+The MIT License (MIT)
+
+Copyright (c) 2013-2014 winlin
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of
+this software and associated documentation files (the "Software"), to deal in
+the Software without restriction, including without limitation the rights to
+use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+the Software, and to permit persons to whom the Software is furnished to do so,
+subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
+COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
+IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+*/
+
+#ifndef SRS_APP_KBPS_HPP
+#define SRS_APP_KBPS_HPP
+
+/*
+#include <srs_app_kbps.hpp>
+*/
+
+#include <srs_core.hpp>
+
+class ISrsProtocolReader;
+class ISrsProtocolWriter;
+
+/**
+* to statistic the kbps of io.
+*/
+class SrsKbps
+{
+private:
+    ISrsProtocolReader* _in;
+    ISrsProtocolWriter* _out;
+public:
+    SrsKbps();
+    virtual ~SrsKbps();
+public:
+    virtual void set_io(ISrsProtocolReader* in, ISrsProtocolWriter* out);
+public:
+    virtual int get_send_kbps();
+    virtual int get_recv_kbps();
+};
+
+#endif

@@ -43,6 +43,7 @@ using namespace std;
 #include <srs_app_socket.hpp>
 #include <srs_app_socket.hpp>
 #include <srs_app_http_hooks.hpp>
 #include <srs_app_http_hooks.hpp>
 #include <srs_app_edge.hpp>
 #include <srs_app_edge.hpp>
+#include <srs_app_kbps.hpp>
 
 
 // when stream is busy, for example, streaming is already
 // when stream is busy, for example, streaming is already
 // publishing, when a new client to request to publish,
 // publishing, when a new client to request to publish,
@@ -71,6 +72,8 @@ SrsRtmpConn::SrsRtmpConn(SrsServer* srs_server, st_netfd_t client_stfd)
     refer = new SrsRefer();
     refer = new SrsRefer();
     bandwidth = new SrsBandwidth();
     bandwidth = new SrsBandwidth();
     duration = 0;
     duration = 0;
+    kbps = new SrsKbps();
+    kbps->set_io(skt, skt);
     
     
     _srs_config->subscribe(this);
     _srs_config->subscribe(this);
 }
 }
@@ -87,6 +90,7 @@ SrsRtmpConn::~SrsRtmpConn()
     srs_freep(skt);
     srs_freep(skt);
     srs_freep(refer);
     srs_freep(refer);
     srs_freep(bandwidth);
     srs_freep(bandwidth);
+    srs_freep(kbps);
 }
 }
 
 
 // TODO: return detail message when error for client.
 // TODO: return detail message when error for client.
@@ -501,9 +505,8 @@ int SrsRtmpConn::playing(SrsSource* source)
         // reportable
         // reportable
         if (pithy_print.can_print()) {
         if (pithy_print.can_print()) {
             srs_trace("-> "SRS_LOG_ID_PLAY
             srs_trace("-> "SRS_LOG_ID_PLAY
-                " time=%"PRId64", duration=%"PRId64", msgs=%d, obytes=%"PRId64", ibytes=%"PRId64", okbps=%d, ikbps=%d", 
-                pithy_print.age(), duration, count, rtmp->get_send_bytes(), rtmp->get_recv_bytes(), 
-                rtmp->get_send_kbps(), rtmp->get_recv_kbps());
+                " time=%"PRId64", duration=%"PRId64", msgs=%d, okbps=%d, ikbps=%d", 
+                pithy_print.age(), duration, count, kbps->get_send_kbps(), kbps->get_recv_kbps());
         }
         }
         
         
         if (count <= 0) {
         if (count <= 0) {
@@ -586,9 +589,8 @@ int SrsRtmpConn::fmle_publish(SrsSource* source)
         // reportable
         // reportable
         if (pithy_print.can_print()) {
         if (pithy_print.can_print()) {
             srs_trace("<- "SRS_LOG_ID_CLIENT_PUBLISH
             srs_trace("<- "SRS_LOG_ID_CLIENT_PUBLISH
-                " time=%"PRId64", obytes=%"PRId64", ibytes=%"PRId64", okbps=%d, ikbps=%d", 
-                pithy_print.age(), rtmp->get_send_bytes(), rtmp->get_recv_bytes(), 
-                rtmp->get_send_kbps(), rtmp->get_recv_kbps());
+                " time=%"PRId64", okbps=%d, ikbps=%d", 
+                pithy_print.age(), kbps->get_send_kbps(), kbps->get_recv_kbps());
         }
         }
     
     
         // process UnPublish event.
         // process UnPublish event.
@@ -663,9 +665,8 @@ int SrsRtmpConn::flash_publish(SrsSource* source)
         // reportable
         // reportable
         if (pithy_print.can_print()) {
         if (pithy_print.can_print()) {
             srs_trace("<- "SRS_LOG_ID_WEB_PUBLISH
             srs_trace("<- "SRS_LOG_ID_WEB_PUBLISH
-                " time=%"PRId64", obytes=%"PRId64", ibytes=%"PRId64", okbps=%d, ikbps=%d", 
-                pithy_print.age(), rtmp->get_send_bytes(), rtmp->get_recv_bytes(), 
-                rtmp->get_send_kbps(), rtmp->get_recv_kbps());
+                " time=%"PRId64", okbps=%d, ikbps=%d", 
+                pithy_print.age(), kbps->get_send_kbps(), kbps->get_recv_kbps());
         }
         }
     
     
         // process UnPublish event.
         // process UnPublish event.

@@ -46,6 +46,7 @@ class SrsSocket;
 class SrsHttpHooks;
 class SrsHttpHooks;
 #endif
 #endif
 class SrsBandwidth;
 class SrsBandwidth;
+class SrsKbps;
 
 
 /**
 /**
 * the client provides the main logic control for RTMP clients.
 * the client provides the main logic control for RTMP clients.
@@ -63,6 +64,7 @@ private:
     // for live play duration, for instance, rtmpdump to record.
     // for live play duration, for instance, rtmpdump to record.
     // @see https://github.com/winlinvip/simple-rtmp-server/issues/47
     // @see https://github.com/winlinvip/simple-rtmp-server/issues/47
     int64_t duration;
     int64_t duration;
+    SrsKbps* kbps;
 public:
 public:
     SrsRtmpConn(SrsServer* srs_server, st_netfd_t client_stfd);
     SrsRtmpConn(SrsServer* srs_server, st_netfd_t client_stfd);
     virtual ~SrsRtmpConn();
     virtual ~SrsRtmpConn();

@@ -31,7 +31,6 @@ SrsSocket::SrsSocket(st_netfd_t client_stfd)
     stfd = client_stfd;
     stfd = client_stfd;
     send_timeout = recv_timeout = ST_UTIME_NO_TIMEOUT;
     send_timeout = recv_timeout = ST_UTIME_NO_TIMEOUT;
     recv_bytes = send_bytes = 0;
     recv_bytes = send_bytes = 0;
-    start_time_ms = srs_get_system_time_ms();
 }
 }
 
 
 SrsSocket::~SrsSocket()
 SrsSocket::~SrsSocket()
@@ -73,28 +72,6 @@ int64_t SrsSocket::get_send_bytes()
     return send_bytes;
     return send_bytes;
 }
 }
 
 
-int SrsSocket::get_recv_kbps()
-{
-    int64_t diff_ms = srs_get_system_time_ms() - start_time_ms;
-    
-    if (diff_ms <= 0) {
-        return 0;
-    }
-    
-    return recv_bytes * 8 / diff_ms;
-}
-
-int SrsSocket::get_send_kbps()
-{
-    int64_t diff_ms = srs_get_system_time_ms() - start_time_ms;
-    
-    if (diff_ms <= 0) {
-        return 0;
-    }
-    
-    return send_bytes * 8 / diff_ms;
-}
-
 int SrsSocket::read(const void* buf, size_t size, ssize_t* nread)
 int SrsSocket::read(const void* buf, size_t size, ssize_t* nread)
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;

@@ -44,7 +44,6 @@ private:
     int64_t send_timeout;
     int64_t send_timeout;
     int64_t recv_bytes;
     int64_t recv_bytes;
     int64_t send_bytes;
     int64_t send_bytes;
-    int64_t start_time_ms;
     st_netfd_t stfd;
     st_netfd_t stfd;
 public:
 public:
     SrsSocket(st_netfd_t client_stfd);
     SrsSocket(st_netfd_t client_stfd);
@@ -57,8 +56,6 @@ public:
     virtual int64_t get_send_timeout();
     virtual int64_t get_send_timeout();
     virtual int64_t get_recv_bytes();
     virtual int64_t get_recv_bytes();
     virtual int64_t get_send_bytes();
     virtual int64_t get_send_bytes();
-    virtual int get_recv_kbps();
-    virtual int get_send_kbps();
 public:
 public:
     /**
     /**
     * @param nread, the actual read bytes, ignore if NULL.
     * @param nread, the actual read bytes, ignore if NULL.

@@ -44,9 +44,6 @@ SimpleSocketStream::SimpleSocketStream()
     fd = -1;
     fd = -1;
     send_timeout = recv_timeout = ST_UTIME_NO_TIMEOUT;
     send_timeout = recv_timeout = ST_UTIME_NO_TIMEOUT;
     recv_bytes = send_bytes = 0;
     recv_bytes = send_bytes = 0;
-    
-    srs_update_system_time_ms();
-    start_time_ms = srs_get_system_time_ms();
 }
 }
 
 
 SimpleSocketStream::~SimpleSocketStream()
 SimpleSocketStream::~SimpleSocketStream()
@@ -122,18 +119,6 @@ int64_t SimpleSocketStream::get_recv_bytes()
     return recv_bytes;
     return recv_bytes;
 }
 }
 
 
-int SimpleSocketStream::get_recv_kbps()
-{
-    srs_update_system_time_ms();
-    int64_t diff_ms = srs_get_system_time_ms() - start_time_ms;
-    
-    if (diff_ms <= 0) {
-        return 0;
-    }
-    
-    return recv_bytes * 8 / diff_ms;
-}
-
 // ISrsProtocolWriter
 // ISrsProtocolWriter
 void SimpleSocketStream::set_send_timeout(int64_t timeout_us)
 void SimpleSocketStream::set_send_timeout(int64_t timeout_us)
 {
 {
@@ -150,18 +135,6 @@ int64_t SimpleSocketStream::get_send_bytes()
     return send_bytes;
     return send_bytes;
 }
 }
 
 
-int SimpleSocketStream::get_send_kbps()
-{
-    srs_update_system_time_ms();
-    int64_t diff_ms = srs_get_system_time_ms() - start_time_ms;
-    
-    if (diff_ms <= 0) {
-        return 0;
-    }
-    
-    return send_bytes * 8 / diff_ms;
-}
-
 int SimpleSocketStream::writev(const iovec *iov, int iov_size, ssize_t* nwrite)
 int SimpleSocketStream::writev(const iovec *iov, int iov_size, ssize_t* nwrite)
 {
 {
     int ret = ERROR_SUCCESS;
     int ret = ERROR_SUCCESS;

@@ -39,7 +39,6 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 class SimpleSocketStream : public ISrsProtocolReaderWriter
 class SimpleSocketStream : public ISrsProtocolReaderWriter
 {
 {
 private:
 private:
-    int64_t start_time_ms;
     int64_t recv_timeout;
     int64_t recv_timeout;
     int64_t send_timeout;
     int64_t send_timeout;
     int64_t recv_bytes;
     int64_t recv_bytes;
@@ -59,13 +58,11 @@ public:
     virtual void set_recv_timeout(int64_t timeout_us);
     virtual void set_recv_timeout(int64_t timeout_us);
     virtual int64_t get_recv_timeout();
     virtual int64_t get_recv_timeout();
     virtual int64_t get_recv_bytes();
     virtual int64_t get_recv_bytes();
-    virtual int get_recv_kbps();
 // ISrsProtocolWriter
 // ISrsProtocolWriter
 public:
 public:
     virtual void set_send_timeout(int64_t timeout_us);
     virtual void set_send_timeout(int64_t timeout_us);
     virtual int64_t get_send_timeout();
     virtual int64_t get_send_timeout();
     virtual int64_t get_send_bytes();
     virtual int64_t get_send_bytes();
-    virtual int get_send_kbps();
     virtual int writev(const iovec *iov, int iov_size, ssize_t* nwrite);
     virtual int writev(const iovec *iov, int iov_size, ssize_t* nwrite);
 // ISrsProtocolReaderWriter
 // ISrsProtocolReaderWriter
 public:
 public:

@@ -47,8 +47,6 @@ public:
     virtual void set_recv_timeout(int64_t timeout_us) = 0;
     virtual void set_recv_timeout(int64_t timeout_us) = 0;
     virtual int64_t get_recv_timeout() = 0;
     virtual int64_t get_recv_timeout() = 0;
     virtual int64_t get_recv_bytes() = 0;
     virtual int64_t get_recv_bytes() = 0;
-    // TODO: FIXME: remove this interface.
-    virtual int get_recv_kbps() = 0;
 };
 };
 
 
 /**
 /**
@@ -64,8 +62,6 @@ public:
     virtual void set_send_timeout(int64_t timeout_us) = 0;
     virtual void set_send_timeout(int64_t timeout_us) = 0;
     virtual int64_t get_send_timeout() = 0;
     virtual int64_t get_send_timeout() = 0;
     virtual int64_t get_send_bytes() = 0;
     virtual int64_t get_send_bytes() = 0;
-    // TODO: FIXME: remove this interface.
-    virtual int get_send_kbps() = 0;
     virtual int writev(const iovec *iov, int iov_size, ssize_t* nwrite) = 0;
     virtual int writev(const iovec *iov, int iov_size, ssize_t* nwrite) = 0;
 };
 };
 
 

@@ -375,16 +375,6 @@ int64_t SrsRtmpClient::get_send_bytes()
     return protocol->get_send_bytes();
     return protocol->get_send_bytes();
 }
 }
 
 
-int SrsRtmpClient::get_recv_kbps()
-{
-    return protocol->get_recv_kbps();
-}
-
-int SrsRtmpClient::get_send_kbps()
-{
-    return protocol->get_send_kbps();
-}
-
 int SrsRtmpClient::recv_message(SrsMessage** pmsg)
 int SrsRtmpClient::recv_message(SrsMessage** pmsg)
 {
 {
     return protocol->recv_message(pmsg);
     return protocol->recv_message(pmsg);
@@ -730,16 +720,6 @@ int64_t SrsRtmpServer::get_send_bytes()
     return protocol->get_send_bytes();
     return protocol->get_send_bytes();
 }
 }
 
 
-int SrsRtmpServer::get_recv_kbps()
-{
-    return protocol->get_recv_kbps();
-}
-
-int SrsRtmpServer::get_send_kbps()
-{
-    return protocol->get_send_kbps();
-}
-
 int SrsRtmpServer::recv_message(SrsMessage** pmsg)
 int SrsRtmpServer::recv_message(SrsMessage** pmsg)
 {
 {
     return protocol->recv_message(pmsg);
     return protocol->recv_message(pmsg);

@@ -161,8 +161,6 @@ public:
     virtual void set_send_timeout(int64_t timeout_us);
     virtual void set_send_timeout(int64_t timeout_us);
     virtual int64_t get_recv_bytes();
     virtual int64_t get_recv_bytes();
     virtual int64_t get_send_bytes();
     virtual int64_t get_send_bytes();
-    virtual int get_recv_kbps();
-    virtual int get_send_kbps();
     virtual int recv_message(SrsMessage** pmsg);
     virtual int recv_message(SrsMessage** pmsg);
     virtual int decode_message(SrsMessage* msg, SrsPacket** ppacket);
     virtual int decode_message(SrsMessage* msg, SrsPacket** ppacket);
     virtual int send_and_free_message(SrsMessage* msg);
     virtual int send_and_free_message(SrsMessage* msg);
@@ -208,8 +206,6 @@ public:
     virtual int64_t get_send_timeout();
     virtual int64_t get_send_timeout();
     virtual int64_t get_recv_bytes();
     virtual int64_t get_recv_bytes();
     virtual int64_t get_send_bytes();
     virtual int64_t get_send_bytes();
-    virtual int get_recv_kbps();
-    virtual int get_send_kbps();
     virtual int recv_message(SrsMessage** pmsg);
     virtual int recv_message(SrsMessage** pmsg);
     virtual int decode_message(SrsMessage* msg, SrsPacket** ppacket);
     virtual int decode_message(SrsMessage* msg, SrsPacket** ppacket);
     virtual int send_and_free_message(SrsMessage* msg);
     virtual int send_and_free_message(SrsMessage* msg);

@@ -359,16 +359,6 @@ int64_t SrsProtocol::get_send_bytes()
     return skt->get_send_bytes();
     return skt->get_send_bytes();
 }
 }
 
 
-int SrsProtocol::get_recv_kbps()
-{
-    return skt->get_recv_kbps();
-}
-
-int SrsProtocol::get_send_kbps()
-{
-    return skt->get_send_kbps();
-}
-
 int SrsProtocol::recv_message(SrsMessage** pmsg)
 int SrsProtocol::recv_message(SrsMessage** pmsg)
 {
 {
     *pmsg = NULL;
     *pmsg = NULL;

@@ -138,8 +138,6 @@ public:
     virtual int64_t get_send_timeout();
     virtual int64_t get_send_timeout();
     virtual int64_t get_recv_bytes();
     virtual int64_t get_recv_bytes();
     virtual int64_t get_send_bytes();
     virtual int64_t get_send_bytes();
-    virtual int get_recv_kbps();
-    virtual int get_send_kbps();
 public:
 public:
     /**
     /**
     * recv a RTMP message, which is bytes oriented.
     * recv a RTMP message, which is bytes oriented.

@@ -71,6 +71,8 @@ file
 	..\app\srs_app_ingest.cpp,
 	..\app\srs_app_ingest.cpp,
 	..\app\srs_app_json.hpp,
 	..\app\srs_app_json.hpp,
 	..\app\srs_app_json.cpp,
 	..\app\srs_app_json.cpp,
+	..\app\srs_app_kbps.hpp,
+	..\app\srs_app_kbps.cpp,
 	..\app\srs_app_log.hpp,
 	..\app\srs_app_log.hpp,
 	..\app\srs_app_log.cpp,
 	..\app\srs_app_log.cpp,
 	..\app\srs_app_refer.hpp,
 	..\app\srs_app_refer.hpp,

@@ -73,11 +73,6 @@ int64_t MockEmptyIO::get_recv_bytes()
     return -1;
     return -1;
 }
 }
 
 
-int MockEmptyIO::get_recv_kbps()
-{
-    return 0;
-}
-
 void MockEmptyIO::set_send_timeout(int64_t /*timeout_us*/)
 void MockEmptyIO::set_send_timeout(int64_t /*timeout_us*/)
 {
 {
 }
 }
@@ -92,11 +87,6 @@ int64_t MockEmptyIO::get_send_bytes()
     return 0;
     return 0;
 }
 }
 
 
-int MockEmptyIO::get_send_kbps()
-{
-    return 0;
-}
-
 int MockEmptyIO::writev(const iovec */*iov*/, int /*iov_size*/, ssize_t* /*nwrite*/)
 int MockEmptyIO::writev(const iovec */*iov*/, int /*iov_size*/, ssize_t* /*nwrite*/)
 {
 {
     return ERROR_SUCCESS;
     return ERROR_SUCCESS;

@@ -53,13 +53,11 @@ public:
     virtual void set_recv_timeout(int64_t timeout_us);
     virtual void set_recv_timeout(int64_t timeout_us);
     virtual int64_t get_recv_timeout();
     virtual int64_t get_recv_timeout();
     virtual int64_t get_recv_bytes();
     virtual int64_t get_recv_bytes();
-    virtual int get_recv_kbps();
 // for protocol
 // for protocol
 public:
 public:
     virtual void set_send_timeout(int64_t timeout_us);
     virtual void set_send_timeout(int64_t timeout_us);
     virtual int64_t get_send_timeout();
     virtual int64_t get_send_timeout();
     virtual int64_t get_send_bytes();
     virtual int64_t get_send_bytes();
-    virtual int get_send_kbps();
     virtual int writev(const iovec *iov, int iov_size, ssize_t* nwrite);
     virtual int writev(const iovec *iov, int iov_size, ssize_t* nwrite);
 // for protocol/amf0/msg-codec
 // for protocol/amf0/msg-codec
 public:
 public: