refine the kbps calc module. 0.9.93

@@ -229,6 +229,7 @@ Supported operating systems and hardware:
 * 2013-10-17, Created.<br/>
 * 2013-10-17, Created.<br/>
 
 
 ## History
 ## History
+* v1.0, 2014-05-12, refine the kbps calc module. 0.9.93
 * v1.0, 2014-05-08, edge support FMS origin server. 0.9.92
 * v1.0, 2014-05-08, edge support FMS origin server. 0.9.92
 * v1.0, 2014-04-28, [1.0 mainline2(0.9.79)](https://github.com/winlinvip/simple-rtmp-server/releases/tag/1.0.mainline2) released. 35255 lines.
 * v1.0, 2014-04-28, [1.0 mainline2(0.9.79)](https://github.com/winlinvip/simple-rtmp-server/releases/tag/1.0.mainline2) released. 35255 lines.
 * v1.0, 2014-04-28, support full edge RTMP server. 0.9.79
 * v1.0, 2014-04-28, support full edge RTMP server. 0.9.79

@@ -26,11 +26,21 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 #include <srs_kernel_error.hpp>
 #include <srs_kernel_error.hpp>
 #include <srs_kernel_log.hpp>
 #include <srs_kernel_log.hpp>
 #include <srs_protocol_io.hpp>
 #include <srs_protocol_io.hpp>
+#include <srs_kernel_utility.hpp>
+
+SrsKbpsSlice::SrsKbpsSlice()
+{
+    io.in = NULL;
+    io.out = NULL;
+    last_bytes = io_bytes_base = starttime = bytes = 0;
+}
+
+SrsKbpsSlice::~SrsKbpsSlice()
+{
+}
 
 
 SrsKbps::SrsKbps()
 SrsKbps::SrsKbps()
 {
 {
-    _in = NULL;
-    _out = NULL;
 }
 }
 
 
 SrsKbps::~SrsKbps()
 SrsKbps::~SrsKbps()
@@ -39,17 +49,72 @@ SrsKbps::~SrsKbps()
 
 
 void SrsKbps::set_io(ISrsProtocolReader* in, ISrsProtocolWriter* out)
 void SrsKbps::set_io(ISrsProtocolReader* in, ISrsProtocolWriter* out)
 {
 {
-    _in = in;
-    _out = out;
+    // set input stream
+    // now, set start time.
+    if (is.starttime == 0) {
+        is.starttime = srs_get_system_time_ms();
+    }
+    // save the old in bytes.
+    if (is.io.in) {
+        is.bytes += is.last_bytes - is.io_bytes_base;
+    }
+    // use new io.
+    is.io.in = in;
+    is.last_bytes = is.io_bytes_base = 0;
+    if (in) {
+        is.last_bytes = is.io_bytes_base = in->get_recv_bytes();
+    }
+    
+    // set output stream
+    // now, set start time.
+    if (os.starttime == 0) {
+        os.starttime = srs_get_system_time_ms();
+    }
+    // save the old in bytes.
+    if (os.io.out) {
+        os.bytes += os.last_bytes - os.io_bytes_base;
+    }
+    // use new io.
+    os.io.out = out;
+    os.last_bytes = os.io_bytes_base = 0;
+    if (out) {
+        os.last_bytes = os.io_bytes_base = out->get_send_bytes();
+    }
 }
 }
 
 
 int SrsKbps::get_send_kbps()
 int SrsKbps::get_send_kbps()
 {
 {
+    int64_t duration = srs_get_system_time_ms() - is.starttime;
+    int64_t bytes = get_send_bytes();
+    if (duration <= 0) {
         return 0;
         return 0;
+    }
+    return bytes * 8 / duration;
 }
 }
 
 
 int SrsKbps::get_recv_kbps()
 int SrsKbps::get_recv_kbps()
 {
 {
+    int64_t duration = srs_get_system_time_ms() - os.starttime;
+    int64_t bytes = get_recv_bytes();
+    if (duration <= 0) {
         return 0;
         return 0;
+    }
+    return bytes * 8 / duration;
+}
+
+int64_t SrsKbps::get_send_bytes()
+{
+    if (os.io.out) {
+        os.last_bytes = os.io.out->get_send_bytes();
+    }
+    return os.bytes + os.last_bytes - os.io_bytes_base;
+}
+
+int64_t SrsKbps::get_recv_bytes()
+{
+    if (is.io.in) {
+        is.last_bytes = is.io.in->get_recv_bytes();
+    }
+    return is.bytes + is.last_bytes - is.io_bytes_base;
 }
 }
 
 

@@ -34,21 +34,63 @@ class ISrsProtocolReader;
 class ISrsProtocolWriter;
 class ISrsProtocolWriter;
 
 
 /**
 /**
+* a slice of kbps statistic, for input or output.
+*/
+class SrsKbpsSlice
+{
+private:
+    union slice_io {
+        ISrsProtocolReader* in;
+        ISrsProtocolWriter* out;
+    };
+public:
+    slice_io io;
+    int64_t bytes;
+    int64_t starttime;
+    // startup bytes number for io when set it,
+    // the base offset of bytes for io.
+    int64_t io_bytes_base;
+    // last updated bytes number,
+    // cache for io maybe freed.
+    int64_t last_bytes;
+public:
+    SrsKbpsSlice();
+    virtual ~SrsKbpsSlice();
+};
+
+/**
 * to statistic the kbps of io.
 * to statistic the kbps of io.
 */
 */
 class SrsKbps
 class SrsKbps
 {
 {
 private:
 private:
-    ISrsProtocolReader* _in;
-    ISrsProtocolWriter* _out;
+    SrsKbpsSlice is;
+    SrsKbpsSlice os;
 public:
 public:
     SrsKbps();
     SrsKbps();
     virtual ~SrsKbps();
     virtual ~SrsKbps();
 public:
 public:
+    /**
+    * set the underlayer reader/writer,
+    * if the io destroied, for instance, the forwarder reconnect,
+    * user must set the io of SrsKbps to NULL to continue to use the kbps object.
+    * @param in the input stream statistic. can be NULL.
+    * @param out the output stream statistic. can be NULL.
+    * @remark if in/out is NULL, use the cached data for kbps.
+    */
     virtual void set_io(ISrsProtocolReader* in, ISrsProtocolWriter* out);
     virtual void set_io(ISrsProtocolReader* in, ISrsProtocolWriter* out);
 public:
 public:
+    /**
+    * get total kbps, duration is from the startup of io.
+    */
     virtual int get_send_kbps();
     virtual int get_send_kbps();
     virtual int get_recv_kbps();
     virtual int get_recv_kbps();
+public:
+    /**
+    * get the total send/recv bytes, from the startup of the oldest io.
+    */
+    virtual int64_t get_send_bytes();
+    virtual int64_t get_recv_bytes();
 };
 };
 
 
 #endif
 #endif

@@ -31,7 +31,7 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
 // current release version
 // current release version
 #define VERSION_MAJOR "0"
 #define VERSION_MAJOR "0"
 #define VERSION_MINOR "9"
 #define VERSION_MINOR "9"
-#define VERSION_REVISION "92"
+#define VERSION_REVISION "93"
 #define RTMP_SIG_SRS_VERSION VERSION_MAJOR"."VERSION_MINOR"."VERSION_REVISION
 #define RTMP_SIG_SRS_VERSION VERSION_MAJOR"."VERSION_MINOR"."VERSION_REVISION
 // server info.
 // server info.
 #define RTMP_SIG_SRS_KEY "srs"
 #define RTMP_SIG_SRS_KEY "srs"