Add network traffic statistics (#4239)

Co-authored-by: xiongguangjie <xiong_panda@163.com>
Co-authored-by: xia-chu <771730766@qq.com>
This commit is contained in:
PioLing
2025-05-02 16:23:25 +08:00
committed by GitHub
parent ab14adb94d
commit 7b1f8fedac
35 changed files with 325 additions and 19 deletions

View File

@@ -426,6 +426,9 @@ Value ToJson(const PusherProxy::Ptr& p) {
item["status"] = p->getStatus();
item["liveSecs"] = p->getLiveSecs();
item["rePublishCount"] = p->getRePublishCount();
item["bytesSpeed"] = (Json::UInt64) p->getSendSpeed();
item["totalBytes"] =(Json::UInt64) p->getSendTotalBytes();
if (auto src = p->getSrc()) {
dumpMediaTuple(src->getMediaTuple(), item["src"]);
}
@@ -439,6 +442,9 @@ Value ToJson(const PlayerProxy::Ptr& p) {
item["liveSecs"] = p->getLiveSecs();
item["rePullCount"] = p->getRePullCount();
item["totalReaderCount"] = p->totalReaderCount();
item["bytesSpeed"] = (Json::UInt64) p->getRecvSpeed();
item["totalBytes"] = (Json::UInt64) p->getRecvTotalBytes();
dumpMediaTuple(p->getMediaTuple(), item["src"]);
return item;
}
@@ -449,7 +455,8 @@ Value makeMediaSourceJson(MediaSource &media){
dumpMediaTuple(media.getMediaTuple(), item);
item["createStamp"] = (Json::UInt64) media.getCreateStamp();
item["aliveSecond"] = (Json::UInt64) media.getAliveSecond();
item["bytesSpeed"] = media.getBytesSpeed();
item["bytesSpeed"] = (Json::UInt64) media.getBytesSpeed();
item["totalBytes"] = (Json::UInt64) media.getTotalBytes();
item["readerCount"] = media.readerCount();
item["totalReaderCount"] = media.totalReaderCount();
item["originType"] = (int) media.getOriginType();
@@ -1674,8 +1681,10 @@ void installWebApi() {
CHECK(muxer, "get muxer from media source failed");
src->getOwnerPoller()->async([=]() mutable {
muxer->forEachRtpSender([&](const std::string &ssrc) mutable {
muxer->forEachRtpSender([&](const std::string &ssrc, const RtpSender &sender) mutable {
val["data"].append(ssrc);
val["bytesSpeed"] = (Json::UInt64)sender.getSendSpeed();
val["totalBytes"] = (Json::UInt64)sender.getSendTotalBytes();
});
invoker(200, headerOut, val.toStyledString());
});

View File

@@ -110,13 +110,20 @@ std::shared_ptr<void> MediaSource::getOwnership() {
});
}
int MediaSource::getBytesSpeed(TrackType type){
size_t MediaSource::getBytesSpeed(TrackType type) {
if(type == TrackInvalid || type == TrackMax){
return _speed[TrackVideo].getSpeed() + _speed[TrackAudio].getSpeed();
}
return _speed[type].getSpeed();
}
size_t MediaSource::getTotalBytes(TrackType type) {
if (type == TrackInvalid || type == TrackMax) {
return _speed[TrackVideo].getTotalBytes() + _speed[TrackAudio].getTotalBytes();
}
return _speed[type].getTotalBytes();
}
uint64_t MediaSource::getAliveSecond() const {
// 使用Ticker对象获取存活时间的目的是防止修改系统时间导致回退 [AUTO-TRANSLATED:68474061]
// The purpose of using the Ticker object to obtain the survival time is to prevent the modification of the system time from causing a rollback

View File

@@ -442,7 +442,9 @@ public:
// 获取数据速率单位bytes/s [AUTO-TRANSLATED:c70465c1]
// Get data rate, unit bytes/s
int getBytesSpeed(TrackType type = TrackInvalid);
size_t getBytesSpeed(TrackType type = TrackInvalid);
size_t getTotalBytes(TrackType type = TrackInvalid);
// 获取流创建GMT unix时间戳单位秒 [AUTO-TRANSLATED:0bbe145e]
// Get the stream creation GMT unix timestamp, unit seconds
uint64_t getCreateStamp() const { return _create_stamp; }

View File

@@ -183,9 +183,12 @@ std::string MultiMediaSourceMuxer::shortUrl() const {
return _tuple.shortUrl();
}
void MultiMediaSourceMuxer::forEachRtpSender(const std::function<void(const std::string &ssrc)> &cb) const {
void MultiMediaSourceMuxer::forEachRtpSender(const std::function<void(const std::string &ssrc, const RtpSender &sender)> &cb) const {
for (auto &pr : _rtp_sender) {
cb(pr.first);
auto sender = std::get<1>(pr.second).lock();
if (sender) {
cb(pr.first, *sender);
}
}
}
@@ -446,7 +449,8 @@ void MultiMediaSourceMuxer::startSendRtp(MediaSource &sender, const MediaSourceE
if (!ssrc_multi_send) {
strong_self->_rtp_sender.erase(ssrc);
}
strong_self->_rtp_sender.emplace(ssrc,reader);
std::weak_ptr<RtpSender> sender = rtp_sender;
strong_self->_rtp_sender.emplace(ssrc, make_tuple(reader, sender));
});
});
#else

View File

@@ -194,7 +194,7 @@ public:
const MediaTuple &getMediaTuple() const;
std::string shortUrl() const;
void forEachRtpSender(const std::function<void(const std::string &ssrc)> &cb) const;
void forEachRtpSender(const std::function<void(const std::string &ssrc, const RtpSender &sender)> &cb) const;
protected:
/////////////////////////////////MediaSink override/////////////////////////////////
@@ -245,7 +245,7 @@ private:
toolkit::Ticker _last_check;
std::unordered_map<int, Stamp> _stamps;
std::weak_ptr<Listener> _track_listener;
std::unordered_multimap<std::string, RingType::RingReader::Ptr> _rtp_sender;
std::unordered_multimap<std::string, std::tuple<RingType::RingReader::Ptr, std::weak_ptr<RtpSender>>> _rtp_sender;
FMP4MediaSourceMuxer::Ptr _fmp4;
RtmpMediaSourceMuxer::Ptr _rtmp;
RtspMediaSourceMuxer::Ptr _rtsp;

View File

@@ -282,6 +282,7 @@ void HlsPlayer::onResponseHeader(const string &status, const HttpClient::HttpHea
void HlsPlayer::onResponseBody(const char *buf, size_t size) {
_m3u8.append(buf, size);
_recvtotalbytes += getRecvTotalBytes();
}
void HlsPlayer::onResponseCompleted(const SockException &ex) {
@@ -357,6 +358,13 @@ void HlsPlayer::playDelay(float delay_sec) {
}, getPoller()));
}
size_t HlsPlayer::getRecvSpeed() {
return TcpClient::getRecvSpeed() + (_http_ts_player ? _http_ts_player->getRecvSpeed() : 0);
}
size_t HlsPlayer::getRecvTotalBytes() {
return TcpClient::getRecvTotalBytes() + (_http_ts_player ? _http_ts_player->getRecvTotalBytes() : 0);
}
//////////////////////////////////////////////////////////////////////////
void HlsDemuxer::start(const EventPoller::Ptr &poller, TrackListener *listener) {
@@ -480,6 +488,7 @@ void HlsPlayerImp::onPacket(const char *data, size_t len) {
if (_decoder && _demuxer) {
_decoder->input((uint8_t *) data, len);
}
_recvtotalbytes += HlsPlayer::getRecvTotalBytes();
}
void HlsPlayerImp::addTrackCompleted() {
@@ -531,4 +540,11 @@ vector<Track::Ptr> HlsPlayerImp::getTracks(bool ready) const {
return static_pointer_cast<HlsDemuxer>(_demuxer)->getTracks(ready);
}
size_t HlsPlayerImp::getRecvSpeed() {
return PlayerImp<HlsPlayer, PlayerBase>::getRecvSpeed();
}
size_t HlsPlayerImp::getRecvTotalBytes() {
return _recvtotalbytes;
}
}//namespace mediakit

View File

@@ -73,6 +73,9 @@ public:
*/
void teardown() override;
size_t getRecvSpeed() override;
size_t getRecvTotalBytes() override;
protected:
/**
* 收到ts包
@@ -127,12 +130,17 @@ private:
int _timeout_multiple = MIN_TIMEOUT_MULTIPLE;
int _try_fetch_index_times = 0;
int _ts_download_failed_count = 0;
protected:
size_t _recvtotalbytes = 0;
};
class HlsPlayerImp : public PlayerImp<HlsPlayer, PlayerBase>, private TrackListener {
class HlsPlayerImp final: public PlayerImp<HlsPlayer, PlayerBase>, private TrackListener {
public:
using Ptr = std::shared_ptr<HlsPlayerImp>;
HlsPlayerImp(const toolkit::EventPoller::Ptr &poller = nullptr);
size_t getRecvSpeed() override;
size_t getRecvTotalBytes() override;
private:
//// HlsPlayer override////

View File

@@ -57,4 +57,11 @@ void TsPlayer::onResponseBody(const char *buf, size_t size) {
}
}
size_t TsPlayer::getRecvSpeed() {
return TcpClient::getRecvSpeed();
}
size_t TsPlayer::getRecvTotalBytes() {
return TcpClient::getRecvTotalBytes();
}
} // namespace mediakit

View File

@@ -28,6 +28,9 @@ public:
*/
void play(const std::string &url) override;
size_t getRecvSpeed() override;
size_t getRecvTotalBytes() override;
/**
* 停止播放
* Stop playing

View File

@@ -21,6 +21,8 @@ public:
using Ptr = std::shared_ptr<TsPlayerImp>;
TsPlayerImp(const toolkit::EventPoller::Ptr &poller = nullptr);
size_t getRecvSpeed() override;
size_t getRecvTotalBytes() override;
private:
//// TsPlayer override////

View File

@@ -77,4 +77,11 @@ vector<Track::Ptr> TsPlayerImp::getTracks(bool ready) const {
return static_pointer_cast<HlsDemuxer>(_demuxer)->getTracks(ready);
}
size_t TsPlayerImp::getRecvSpeed() {
return TcpClient::getRecvSpeed();
}
size_t TsPlayerImp::getRecvTotalBytes() {
return TcpClient::getRecvTotalBytes();
}
}//namespace mediakit

View File

@@ -166,6 +166,10 @@ public:
*/
virtual void setOnResume(const std::function<void()> &cb) = 0;
virtual size_t getRecvSpeed() { return 0; }
virtual size_t getRecvTotalBytes() { return 0; }
protected:
virtual void onResume() = 0;
virtual void onShutdown(const toolkit::SockException &ex) = 0;
@@ -256,6 +260,14 @@ public:
_on_resume = cb;
}
size_t getRecvSpeed() override {
return _delegate ? _delegate->getRecvSpeed() : Parent::getRecvSpeed();
}
size_t getRecvTotalBytes() override {
return _delegate ? _delegate->getRecvTotalBytes() : Parent::getRecvTotalBytes();
}
protected:
void onShutdown(const toolkit::SockException &ex) override {
if (_on_shutdown) {

View File

@@ -35,6 +35,7 @@ public:
void setOnCreateSocket(toolkit::Socket::onCreateSocket cb);
std::shared_ptr<MediaSource> getSrc() { return _src.lock(); }
const std::string& getUrl() const { return _url; }
private:
std::weak_ptr<MediaSource> _src;
toolkit::EventPoller::Ptr _poller;

View File

@@ -67,6 +67,9 @@ public:
*/
virtual void setOnShutdown(const Event &cb) = 0;
virtual size_t getSendSpeed() { return 0; }
virtual size_t getSendTotalBytes() { return 0; }
protected:
virtual void onShutdown(const toolkit::SockException &ex) = 0;
virtual void onPublishResult(const toolkit::SockException &ex) = 0;
@@ -133,6 +136,14 @@ public:
_on_shutdown = cb;
}
size_t getSendSpeed() override {
return _delegate ? _delegate->getSendSpeed() : Parent::getSendSpeed();
}
size_t getSendTotalBytes() override {
return _delegate ? _delegate->getSendTotalBytes() : Parent::getSendTotalBytes();
}
protected:
void onShutdown(const toolkit::SockException &ex) override {
if (_on_shutdown) {

View File

@@ -76,4 +76,12 @@ void FlvPlayer::onRecvRtmpPacket(RtmpPacket::Ptr packet) {
onRtmpPacket(std::move(packet));
}
size_t FlvPlayer::getRecvSpeed() {
return TcpClient::getRecvSpeed();
}
size_t FlvPlayer::getRecvTotalBytes() {
return TcpClient::getRecvTotalBytes();
}
}//mediakit

View File

@@ -23,6 +23,8 @@ public:
void play(const std::string &url) override;
void teardown() override;
size_t getRecvSpeed() override;
size_t getRecvTotalBytes() override;
protected:
void onResponseHeader(const std::string &status, const HttpHeader &header) override;

View File

@@ -452,4 +452,12 @@ void RtmpPlayer::seekToMilliSecond(uint32_t seekMS){
});
}
size_t RtmpPlayer::getRecvSpeed() {
return TcpClient::getRecvSpeed();
}
size_t RtmpPlayer::getRecvTotalBytes() {
return TcpClient::getRecvTotalBytes();
}
} /* namespace mediakit */

View File

@@ -37,6 +37,9 @@ public:
void speed(float speed) override;
void teardown() override;
size_t getRecvSpeed() override;
size_t getRecvTotalBytes() override;
protected:
virtual bool onMetadata(const AMFValue &val) = 0;
virtual void onRtmpPacket(RtmpPacket::Ptr chunk_data) = 0;

View File

@@ -332,6 +332,12 @@ void RtmpPusher::onRtmpChunk(RtmpPacket::Ptr packet) {
}
}
size_t RtmpPusher::getSendSpeed() {
return TcpClient::getSendSpeed();
}
size_t RtmpPusher::getSendTotalBytes() {
return TcpClient::getSendTotalBytes();
}
} /* namespace mediakit */

View File

@@ -27,6 +27,9 @@ public:
void publish(const std::string &url) override ;
void teardown() override;
size_t getSendSpeed() override;
size_t getSendTotalBytes() override;
protected:
//for Tcpclient override
void onRecv(const toolkit::Buffer::Ptr &buf) override;

View File

@@ -13,7 +13,7 @@
#include "RawEncoder.h"
#include "Extension/Factory.h"
#include "Rtsp/RtspMuxer.h"
#include "Common//config.h"
#include "Common/config.h"
using namespace toolkit;

View File

@@ -457,5 +457,49 @@ void RtpSender::setOnClose(std::function<void(const toolkit::SockException &ex)>
_on_close = std::move(on_close);
}
size_t RtpSender::getSendSpeed() const {
size_t ret = 0;
if (_socket_rtp) {
ret += _socket_rtp->getSendSpeed();
}
if (_socket_rtcp) {
ret += _socket_rtcp->getSendSpeed();
}
return ret;
}
size_t RtpSender::getRecvSpeed() const {
size_t ret = 0;
if (_socket_rtp) {
ret += _socket_rtp->getRecvSpeed();
}
if (_socket_rtcp) {
ret += _socket_rtcp->getRecvSpeed();
}
return ret;
}
size_t RtpSender::getRecvTotalBytes() const {
size_t ret = 0;
if (_socket_rtp) {
ret += _socket_rtp->getRecvTotalBytes();
}
if (_socket_rtcp) {
ret += _socket_rtcp->getRecvTotalBytes();
}
return ret;
}
size_t RtpSender::getSendTotalBytes() const {
size_t ret = 0;
if (_socket_rtp) {
ret += _socket_rtp->getSendTotalBytes();
}
if (_socket_rtcp) {
ret += _socket_rtcp->getSendTotalBytes();
}
return ret;
}
} // namespace mediakit
#endif // defined(ENABLE_RTPPROXY)

View File

@@ -94,6 +94,11 @@ public:
*/
void setOnClose(std::function<void(const toolkit::SockException &ex)> on_close);
size_t getSendSpeed() const;
size_t getRecvSpeed() const;
size_t getRecvTotalBytes() const;
size_t getSendTotalBytes() const;
private:
// 合并写输出 [AUTO-TRANSLATED:23544836]
// Merge write output

View File

@@ -800,7 +800,7 @@ void RtspPlayer::onPlayResult_l(const SockException &ex, bool handshake_done) {
};
// 创建rtp数据接收超时检测定时器 [AUTO-TRANSLATED:edbffc19]
// Create RTP data receive timeout detection timer
_rtp_check_timer = std::make_shared<Timer>(timeoutMS / 2000.0f, lam, getPoller());
_rtp_check_timer = std::make_shared<Timer>(timeoutMS / 2000.0f, std::move(lam), getPoller());
} else {
sendTeardown();
}
@@ -844,6 +844,36 @@ int RtspPlayer::getTrackIndexByTrackType(TrackType track_type) const {
throw SockException(Err_other, StrPrinter << "no such track with type:" << getTrackString(track_type));
}
size_t RtspPlayer::getRecvSpeed() {
size_t ret = TcpClient::getRecvSpeed();
for (auto &rtp : _rtp_sock) {
if (rtp) {
ret += rtp->getRecvSpeed();
}
}
for (auto &rtcp : _rtcp_sock) {
if (rtcp) {
ret += rtcp->getRecvSpeed();
}
}
return ret;
}
size_t RtspPlayer::getRecvTotalBytes() {
size_t ret = TcpClient::getRecvTotalBytes();
for (auto &rtp : _rtp_sock) {
if (rtp) {
ret += rtp->getRecvTotalBytes();
}
}
for (auto &rtcp : _rtcp_sock) {
if (rtcp) {
ret += rtcp->getRecvTotalBytes();
}
}
return ret;
}
///////////////////////////////////////////////////
// RtspPlayerImp
float RtspPlayerImp::getDuration() const {

View File

@@ -39,6 +39,9 @@ public:
void teardown() override;
float getPacketLossRate(TrackType type) const override;
size_t getRecvSpeed() override;
size_t getRecvTotalBytes() override;
protected:
// 派生类回调函数 [AUTO-TRANSLATED:61e20903]
// Derived class callback function
@@ -159,6 +162,7 @@ private:
float _speed = 0.0f;
std::vector<SdpTrack::Ptr> _sdp_track;
std::function<void(const Parser&)> _on_response;
protected:
// RTP端口,trackid idx 为数组下标 [AUTO-TRANSLATED:77c186bb]
// RTP port, trackid idx is the array subscript
toolkit::Socket::Ptr _rtp_sock[2];
@@ -166,6 +170,7 @@ private:
// RTCP port, trackid idx is the array subscript
toolkit::Socket::Ptr _rtcp_sock[2];
private:
// rtsp鉴权相关 [AUTO-TRANSLATED:947dc6a3]
// Rtsp authentication related
std::string _md5_nonce;
@@ -175,8 +180,10 @@ private:
uint32_t _cseq_send = 1;
std::string _content_base;
std::string _control_url;
protected:
Rtsp::eRtpType _rtp_type = Rtsp::RTP_TCP;
private:
// 当前rtp时间戳 [AUTO-TRANSLATED:410f2691]
// Current rtp timestamp
uint32_t _stamp[2] = {0, 0};

View File

@@ -59,6 +59,36 @@ public:
std::vector<Track::Ptr> getTracks(bool ready = true) const override;
size_t getRecvSpeed() override {
size_t ret = TcpClient::getRecvSpeed();
for (auto &rtp : _rtp_sock) {
if (rtp) {
ret += rtp->getRecvSpeed();
}
}
for (auto &rtcp : _rtcp_sock) {
if (rtcp) {
ret += rtcp->getRecvSpeed();
}
}
return ret;
}
size_t getRecvTotalBytes() override {
size_t ret = TcpClient::getRecvTotalBytes();
for (auto &rtp : _rtp_sock) {
if (rtp) {
ret += rtp->getRecvTotalBytes();
}
}
for (auto &rtcp : _rtcp_sock) {
if (rtcp) {
ret += rtcp->getRecvTotalBytes();
}
}
return ret;
}
private:
// 派生类回调函数 [AUTO-TRANSLATED:61e20903]
// Derived class callback function

View File

@@ -595,5 +595,34 @@ void RtspPusher::sendRtspRequest(const string &cmd, const string &url,const StrC
SockSender::send(std::move(printer));
}
size_t RtspPusher::getSendSpeed() {
size_t ret = TcpClient::getSendSpeed();
for (auto &rtp : _rtp_sock) {
if (rtp) {
ret += rtp->getSendSpeed();
}
}
for (auto &rtcp : _rtcp_sock) {
if (rtcp) {
ret += rtcp->getSendSpeed();
}
}
return ret;
}
size_t RtspPusher::getSendTotalBytes() {
size_t ret = TcpClient::getSendTotalBytes();
for (auto &rtp : _rtp_sock) {
if (rtp) {
ret += rtp->getSendTotalBytes();
}
}
for (auto &rtcp : _rtcp_sock) {
if (rtcp) {
ret += rtcp->getSendTotalBytes();
}
}
return ret;
}
} /* namespace mediakit */

View File

@@ -30,6 +30,8 @@ public:
~RtspPusher() override;
void publish(const std::string &url) override;
void teardown() override;
size_t getSendSpeed() override;
size_t getSendTotalBytes() override;
protected:
//for Tcpclient override

View File

@@ -1018,6 +1018,21 @@ size_t SrtCaller::getPayloadSize() {
return ret;
}
size_t SrtCaller::getRecvSpeed() const {
return _socket ? _socket->getRecvSpeed() : 0;
}
size_t SrtCaller::getRecvTotalBytes() const {
return _socket ? _socket->getRecvTotalBytes() : 0;
}
size_t SrtCaller::getSendSpeed() const {
return _socket ? _socket->getSendSpeed() : 0;
}
size_t SrtCaller::getSendTotalBytes() const {
return _socket ? _socket->getSendTotalBytes() : 0;
}
} /* namespace mediakit */

View File

@@ -60,6 +60,11 @@ public:
virtual void inputSockData(uint8_t *buf, int len, struct sockaddr *addr);
virtual void onSendTSData(const SRT::Buffer::Ptr &buffer, bool flush);
size_t getRecvSpeed() const;
size_t getRecvTotalBytes() const;
size_t getSendSpeed() const;
size_t getSendTotalBytes() const;
protected:
virtual void onConnect();

View File

@@ -120,6 +120,14 @@ std::string SrtPlayer::getPassphrase() {
return passPhrase;
}
size_t SrtPlayer::getRecvSpeed() {
return SrtCaller::getRecvSpeed();
}
size_t SrtPlayer::getRecvTotalBytes() {
return SrtCaller::getRecvTotalBytes();
}
///////////////////////////////////////////////////
// SrtPlayerImp
@@ -164,6 +172,5 @@ void SrtPlayerImp::onSRTData(SRT::DataPacket::Ptr pkt) {
return;
}
} /* namespace mediakit */

View File

@@ -38,6 +38,8 @@ public:
void teardown() override;
void pause(bool pause) override;
void speed(float speed) override;
size_t getRecvSpeed() override;
size_t getRecvTotalBytes() override;
protected:

View File

@@ -112,5 +112,13 @@ void SrtPusher::doPublish() {
});
}
size_t SrtPusher::getSendSpeed() {
return SrtCaller::getSendSpeed();
}
size_t SrtPusher::getSendTotalBytes() {
return SrtCaller::getSendTotalBytes();
}
} /* namespace mediakit */

View File

@@ -51,6 +51,9 @@ protected:
protected:
std::weak_ptr<TSMediaSource> _push_src;
TSMediaSource::RingType::RingReader::Ptr _ts_reader;
size_t getSendSpeed() override;
size_t getSendTotalBytes() override;
};
using SrtPusherImp = PusherImp<SrtPusher, PusherBase>;