添加网速统计功能

server/WebApi.cpp

@@ -395,6 +395,7 @@ void installWebApi() {
         item["vhost"] = media->getVhost();
         item["vhost"] = media->getVhost();
         item["app"] = media->getApp();
         item["app"] = media->getApp();
         item["stream"] = media->getId();
         item["stream"] = media->getId();
+        item["bytes_speed"] = media->getBytesSpeed();
         item["readerCount"] = media->readerCount();
         item["readerCount"] = media->readerCount();
         item["totalReaderCount"] = media->totalReaderCount();
         item["totalReaderCount"] = media->totalReaderCount();
         item["originType"] = (int) media->getOriginType();
         item["originType"] = (int) media->getOriginType();

src/Common/MediaSource.cpp

@@ -66,6 +66,10 @@ const string& MediaSource::getId() const {
     return _stream_id;
     return _stream_id;
 }
 }
 
 
+int MediaSource::getBytesSpeed(){
+    return _speed.getSpeed();
+}
+
 vector<Track::Ptr> MediaSource::getTracks(bool ready) const {
 vector<Track::Ptr> MediaSource::getTracks(bool ready) const {
     auto listener = _listener.lock();
     auto listener = _listener.lock();
     if(!listener){
     if(!listener){

src/Common/MediaSource.h

@@ -137,6 +137,52 @@ public:
     string _param_strs;
     string _param_strs;
 };
 };
 
 
+class BytesSpeed {
+public:
+    BytesSpeed() = default;
+    ~BytesSpeed() = default;
+
+    /**
+     * 添加统计字节
+     */
+    BytesSpeed& operator += (uint64_t bytes) {
+        _bytes += bytes;
+        if (_bytes > 1024 * 1024) {
+            //数据大于1MB就计算一次网速
+            computeSpeed();
+        }
+        return *this;
+    }
+
+    /**
+     * 获取速度，单位bytes/s
+     */
+    int getSpeed() {
+        if (_ticker.elapsedTime() < 1000) {
+            //获取频率小于1秒，那么返回上次计算结果
+            return _speed;
+        }
+        return computeSpeed();
+    }
+
+private:
+    uint64_t computeSpeed() {
+        auto elapsed = _ticker.elapsedTime();
+        if (!elapsed) {
+            return _speed;
+        }
+        _speed = _bytes * 1000 / elapsed;
+        _ticker.resetTime();
+        _bytes = 0;
+        return _speed;
+    }
+
+private:
+    int _speed = 0;
+    uint64_t _bytes = 0;
+    Ticker _ticker;
+};
+
 /**
 /**
  * 媒体源，任何rtsp/rtmp的直播流都源自该对象
  * 媒体源，任何rtsp/rtmp的直播流都源自该对象
  */
  */
@@ -170,6 +216,9 @@ public:
     // 设置时间戳
     // 设置时间戳
     virtual void setTimeStamp(uint32_t stamp) {};
     virtual void setTimeStamp(uint32_t stamp) {};
 
 
+    // 获取数据速率，单位bytes/s
+    int getBytesSpeed();
+
     ////////////////MediaSourceEvent相关接口实现////////////////
     ////////////////MediaSourceEvent相关接口实现////////////////
 
 
     // 设置监听者
     // 设置监听者
@@ -229,6 +278,9 @@ private:
     //触发媒体事件
     //触发媒体事件
     void emitEvent(bool regist);
     void emitEvent(bool regist);
 
 
+protected:
+    BytesSpeed _speed;
+
 private:
 private:
     string _schema;
     string _schema;
     string _vhost;
     string _vhost;

src/FMP4/FMP4MediaSource.h

@@ -99,6 +99,7 @@ public:
         if (key) {
         if (key) {
             _have_video = true;
             _have_video = true;
         }
         }
+        _speed += packet->size();
         PacketCache<FMP4Packet, FMP4FlushPolicy>::inputPacket(true, packet, key);
         PacketCache<FMP4Packet, FMP4FlushPolicy>::inputPacket(true, packet, key);
     }
     }
 
 

src/Record/HlsMakerImp.cpp

@@ -92,6 +92,9 @@ void HlsMakerImp::onWriteSegment(const char *data, int len) {
     if (_file) {
     if (_file) {
         fwrite(data, len, 1, _file.get());
         fwrite(data, len, 1, _file.get());
     }
     }
+    if (_media_src) {
+        _media_src->onSegmentSize(len);
+    }
 }
 }
 
 
 void HlsMakerImp::onWriteHls(const char *data, int len) {
 void HlsMakerImp::onWriteHls(const char *data, int len) {

src/Record/HlsMediaSource.h

@@ -79,6 +79,10 @@ public:
         _list_cb.emplace_back(std::move(cb));
         _list_cb.emplace_back(std::move(cb));
     }
     }
 
 
+    void onSegmentSize(uint64_t bytes) {
+        _speed += bytes;
+    }
+
 private:
 private:
     bool _is_regist = false;
     bool _is_regist = false;
     RingType::Ptr _ring;
     RingType::Ptr _ring;

src/Rtmp/RtmpMediaSource.h

@@ -119,6 +119,7 @@ public:
      * @param pkt rtmp包
      * @param pkt rtmp包
      */
      */
     void onWrite(const RtmpPacket::Ptr &pkt, bool = true) override {
     void onWrite(const RtmpPacket::Ptr &pkt, bool = true) override {
+        _speed += pkt->size();
         //保存当前时间戳
         //保存当前时间戳
         switch (pkt->type_id) {
         switch (pkt->type_id) {
             case MSG_VIDEO : _track_stamps[TrackVideo] = pkt->time_stamp, _have_video = true; break;
             case MSG_VIDEO : _track_stamps[TrackVideo] = pkt->time_stamp, _have_video = true; break;

src/Rtsp/RtspMediaSource.h

@@ -157,6 +157,7 @@ public:
      * @param keyPos 该包是否为关键帧的第一个包
      * @param keyPos 该包是否为关键帧的第一个包
      */
      */
     void onWrite(const RtpPacket::Ptr &rtp, bool keyPos) override {
     void onWrite(const RtpPacket::Ptr &rtp, bool keyPos) override {
+        _speed += rtp->size();
         assert(rtp->type >= 0 && rtp->type < TrackMax);
         assert(rtp->type >= 0 && rtp->type < TrackMax);
         auto track = _tracks[rtp->type];
         auto track = _tracks[rtp->type];
         if (track) {
         if (track) {

src/TS/TSMediaSource.h

@@ -76,6 +76,7 @@ public:
      * @param key 是否为关键帧第一个包
      * @param key 是否为关键帧第一个包
      */
      */
     void onWrite(const TSPacket::Ptr &packet, bool key) override {
     void onWrite(const TSPacket::Ptr &packet, bool key) override {
+        _speed += packet->size();
         if (!_ring) {
         if (!_ring) {
             createRing();
             createRing();
         }
         }