From 7c89c1655f325250cd8aaf3df2a49122d4ae2c26 Mon Sep 17 00:00:00 2001 From: xiongguangjie Date: Wed, 3 May 2023 18:52:11 +0800 Subject: [PATCH] add getProxyPusherInfo and getProxyInfo restful api --- postman/ZLMediaKit.postman_collection.json | 72 +++++++++++++++++++--- server/WebApi.cpp | 42 +++++++++++++ src/Player/PlayerProxy.cpp | 34 +++++++++- src/Player/PlayerProxy.h | 12 ++++ src/Pusher/PusherProxy.cpp | 30 +++++++++ src/Pusher/PusherProxy.h | 9 +++ 6 files changed, 191 insertions(+), 8 deletions(-) diff --git a/postman/ZLMediaKit.postman_collection.json b/postman/ZLMediaKit.postman_collection.json index ff4305ce..044461e0 100644 --- a/postman/ZLMediaKit.postman_collection.json +++ b/postman/ZLMediaKit.postman_collection.json @@ -1,6 +1,6 @@ { "info": { - "_postman_id": "4626d766-16b5-4255-89ba-f7614de2398c", + "_postman_id": "39e8a1df-cc8e-4e3f-bf5e-197c86e7bf0f", "name": "ZLMediaKit", "description": "媒体服务器", "schema": "https://schema.getpostman.com/json/collection/v2.1.0/collection.json" @@ -627,7 +627,7 @@ "method": "GET", "header": [], "url": { - "raw": "{{ZLMediaKit_URL}}/index/api/addStreamPusherProxy?secret={{ZLMediaKit_secret}}&schema=rtmp&vhost={{defaultVhost}}&app=live&stream=test&dst_url=rtmp://127.0.0.1/live/push", + "raw": "{{ZLMediaKit_URL}}/index/api/addStreamPusherProxy?secret={{ZLMediaKit_secret}}&schema=rtmp&vhost={{defaultVhost}}&app=live&stream=test&dst_url=rtmp://192.168.1.104/live/push", "host": [ "{{ZLMediaKit_URL}}" ], @@ -664,7 +664,7 @@ }, { "key": "dst_url", - "value": "rtmp://127.0.0.1/live/push", + "value": "rtmp://192.168.1.104/live/push", "description": "推流地址，需要与schema字段协议一致" }, { @@ -696,7 +696,7 @@ "method": "GET", "header": [], "url": { - "raw": "{{ZLMediaKit_URL}}/index/api/delStreamPusherProxy?secret={{ZLMediaKit_secret}}&key=__defaultVhost__/live/test", + "raw": "{{ZLMediaKit_URL}}/index/api/delStreamPusherProxy?secret={{ZLMediaKit_secret}}&key=rtmp/__defaultVhost__/live/test/f40a8ab006cac16ecc0858409e890491", "host": [ "{{ZLMediaKit_URL}}" ], @@ -713,7 +713,7 @@ }, { "key": "key", - "value": "__defaultVhost__/live/test", + "value": "rtmp/__defaultVhost__/live/test/f40a8ab006cac16ecc0858409e890491", "description": "addStreamPusherProxy接口返回的key" } ] @@ -1363,7 +1363,7 @@ "method": "GET", "header": [], "url": { - "raw": "{{ZLMediaKit_URL}}/index/api/openRtpServer?secret={{ZLMediaKit_secret}}&port=0&enable_tcp=1&stream_id=test", + "raw": "{{ZLMediaKit_URL}}/index/api/openRtpServer?secret={{ZLMediaKit_secret}}&port=0&tcp_mode=1&stream_id=test", "host": [ "{{ZLMediaKit_URL}}" ], @@ -1422,7 +1422,7 @@ "method": "GET", "header": [], "url": { - "raw": "{{ZLMediaKit_URL}}/index/api/connectRtpServer?secret={{ZLMediaKit_secret}}&dst_url=127.0.0.1&dst_port=10000&stream_id=test", + "raw": "{{ZLMediaKit_URL}}/index/api/connectRtpServer?secret={{ZLMediaKit_secret}}&dst_url=0&dst_port=1&stream_id=test", "host": [ "{{ZLMediaKit_URL}}" ], @@ -1874,6 +1874,64 @@ } }, "response": [] + }, + { + "name": "获取拉流代理信息(getProxyInfo)", + "request": { + "method": "GET", + "header": [], + "url": { + "raw": "{{ZLMediaKit_URL}}/index/api/getProxyInfo?secret={{ZLMediaKit_secret}}&key=__defaultVhost__/live/test", + "host": [ + "{{ZLMediaKit_URL}}" + ], + "path": [ + "index", + "api", + "getProxyInfo" + ], + "query": [ + { + "key": "secret", + "value": "{{ZLMediaKit_secret}}" + }, + { + "key": "key", + "value": "__defaultVhost__/live/test" + } + ] + } + }, + "response": [] + }, + { + "name": "获取推流代理信息(getProxyPusherInfo)", + "request": { + "method": "GET", + "header": [], + "url": { + "raw": "{{ZLMediaKit_URL}}/index/api/getProxyPusherInfo?secret={{ZLMediaKit_secret}}&key=rtmp/__defaultVhost__/live/test/f40a8ab006cac16ecc0858409e890491", + "host": [ + "{{ZLMediaKit_URL}}" + ], + "path": [ + "index", + "api", + "getProxyPusherInfo" + ], + "query": [ + { + "key": "secret", + "value": "{{ZLMediaKit_secret}}" + }, + { + "key": "key", + "value": "rtmp/__defaultVhost__/live/test/f40a8ab006cac16ecc0858409e890491" + } + ] + } + }, + "response": [] } ], "event": [ diff --git a/server/WebApi.cpp b/server/WebApi.cpp index bc170cd0..6a1c47e5 100755 --- a/server/WebApi.cpp +++ b/server/WebApi.cpp @@ -1420,6 +1420,48 @@ void installWebApi() { }); }); + api_regist("/index/api/getProxyPusherInfo", [](API_ARGS_MAP_ASYNC) { + CHECK_SECRET(); + CHECK_ARGS("key"); + decltype(s_proxyPusherMap.end()) it; + { + lock_guard lck(s_proxyPusherMapMtx); + it = s_proxyPusherMap.find(allArgs["key"]); + } + + if (it == s_proxyPusherMap.end()) { + throw ApiRetException("can not find pusher", API::NotFound); + } + + auto pusher = it->second; + + val["data"]["status"] = pusher->getStatus(); + val["data"]["liveSecs"] = pusher->getLiveSecs(); + val["data"]["rePublishCount"] = pusher->getRePublishCount(); + invoker(200, headerOut, val.toStyledString()); + }); + + api_regist("/index/api/getProxyInfo", [](API_ARGS_MAP_ASYNC) { + CHECK_SECRET(); + CHECK_ARGS("key"); + decltype(s_proxyMap.end()) it; + { + lock_guard lck(s_proxyMapMtx); + it = s_proxyMap.find(allArgs["key"]); + } + + if (it == s_proxyMap.end()) { + throw ApiRetException("can not find the proxy", API::NotFound); + } + + auto proxy = it->second; + + val["data"]["status"] = proxy->getStatus(); + val["data"]["liveSecs"] = proxy->getLiveSecs(); + val["data"]["rePullCount"] = proxy->getRePullCount(); + invoker(200, headerOut, val.toStyledString()); + }); + // 删除录像文件夹 // http://127.0.0.1/index/api/deleteRecordDirectroy?vhost=__defaultVhost__&app=live&stream=ss&period=2020-01-01 api_regist("/index/api/deleteRecordDirectory", [](API_ARGS_MAP) { diff --git a/src/Player/PlayerProxy.cpp b/src/Player/PlayerProxy.cpp index 521c38e8..ec865788 100644 --- a/src/Player/PlayerProxy.cpp +++ b/src/Player/PlayerProxy.cpp @@ -30,6 +30,10 @@ PlayerProxy::PlayerProxy(const string &vhost, const string &app, const string &s _app = app; _stream_id = stream_id; _retry_count = retry_count; + + _live_secs = 0; + _live_status = 1; + _repull_count = 0; _on_close = [](const SockException &) {}; (*this)[Client::kWaitTrackReady] = false; } @@ -58,10 +62,14 @@ void PlayerProxy::play(const string &strUrlTmp) { if (!err) { // 取消定时器,避免hls拉流索引文件因为网络波动失败重连成功后出现循环重试的情况 - strongSelf->_timer.reset(); + strongSelf->_timer.reset(); + strongSelf->_live_ticker.resetTime(); + strongSelf->_live_status = 0; // 播放成功 *piFailedCnt = 0;//连续播放失败次数清0 strongSelf->onPlaySuccess(); + + InfoL << "play " << strUrlTmp << " success"; } else if (*piFailedCnt < strongSelf->_retry_count || strongSelf->_retry_count < 0) { // 播放失败，延时重试播放 strongSelf->rePlay(strUrlTmp, (*piFailedCnt)++); @@ -92,8 +100,17 @@ void PlayerProxy::play(const string &strUrlTmp) { strongSelf->_muxer->resetTracks(); } } + + if(*piFailedCnt == 0){ + // 第一次重拉更新时长 + strongSelf->_live_secs += strongSelf->_live_ticker.elapsedTime()/1000; + strongSelf->_live_ticker.resetTime(); + TraceL<<" live secs "<_live_secs; + } + //播放异常中断，延时重试播放 if (*piFailedCnt < strongSelf->_retry_count || strongSelf->_retry_count < 0) { + strongSelf->_repull_count++; strongSelf->rePlay(strUrlTmp, (*piFailedCnt)++); } else { //达到了最大重试次数，回调关闭 @@ -241,4 +258,19 @@ void PlayerProxy::onPlaySuccess() { } } +int PlayerProxy::getStatus() { + return _live_status.load(); +} +uint64_t PlayerProxy::getLiveSecs() { + if(_live_status == 0){ + return _live_secs + _live_ticker.elapsedTime()/1000; + }else{ + return _live_secs; + } +} + +uint64_t PlayerProxy::getRePullCount(){ + return _repull_count; +} + } /* namespace mediakit */ diff --git a/src/Player/PlayerProxy.h b/src/Player/PlayerProxy.h index 02cba77b..5ab23481 100644 --- a/src/Player/PlayerProxy.h +++ b/src/Player/PlayerProxy.h @@ -52,6 +52,11 @@ public: */ int totalReaderCount() ; + + int getStatus(); + uint64_t getLiveSecs(); + uint64_t getRePullCount(); + private: //MediaSourceEvent override bool close(MediaSource &sender) override; @@ -76,6 +81,13 @@ private: std::function _on_close; std::function _on_play; MultiMediaSourceMuxer::Ptr _muxer; + + toolkit::Ticker _live_ticker; + //0 表示正常 1 表示正在尝试拉流 + std::atomic _live_status; + std::atomic _live_secs; + + std::atomic _repull_count; }; } /* namespace mediakit */ diff --git a/src/Pusher/PusherProxy.cpp b/src/Pusher/PusherProxy.cpp index 22a658e7..b936842a 100644 --- a/src/Pusher/PusherProxy.cpp +++ b/src/Pusher/PusherProxy.cpp @@ -20,6 +20,9 @@ PusherProxy::PusherProxy(const MediaSource::Ptr &src, int retry_count, const Eve _retry_count = retry_count; _on_close = [](const SockException &) {}; _weak_src = src; + _live_secs = 0; + _live_status = 1; + _republish_count = 0; } PusherProxy::~PusherProxy() { @@ -52,10 +55,14 @@ void PusherProxy::publish(const string &dst_url) { auto src = strong_self->_weak_src.lock(); if (!err) { // 推流成功 + strong_self->_live_ticker.resetTime(); + strong_self->_live_status = 0; *failed_cnt = 0; InfoL << "Publish " << dst_url << " success"; } else if (src && (*failed_cnt < strong_self->_retry_count || strong_self->_retry_count < 0)) { // 推流失败，延时重试推送 + strong_self->_republish_count++; + strong_self->_live_status = 1; strong_self->rePublish(dst_url, (*failed_cnt)++); } else { //如果媒体源已经注销, 或达到了最大重试次数，回调关闭 @@ -69,9 +76,17 @@ void PusherProxy::publish(const string &dst_url) { return; } + if(*failed_cnt == 0){ + // 第一次重推更新时长 + strong_self->_live_secs += strong_self->_live_ticker.elapsedTime()/1000; + strong_self->_live_ticker.resetTime(); + TraceL<<" live secs "<_live_secs; + } + auto src = strong_self->_weak_src.lock(); //推流异常中断，延时重试播放 if (src && (*failed_cnt < strong_self->_retry_count || strong_self->_retry_count < 0)) { + strong_self->_republish_count++; strong_self->rePublish(dst_url, (*failed_cnt)++); } else { //如果媒体源已经注销, 或达到了最大重试次数，回调关闭 @@ -97,4 +112,19 @@ void PusherProxy::rePublish(const string &dst_url, int failed_cnt) { }, getPoller()); } +int PusherProxy::getStatus() { + return _live_status.load(); +} +uint64_t PusherProxy::getLiveSecs() { + if(_live_status == 0){ + return _live_secs + _live_ticker.elapsedTime()/1000; + }else{ + return _live_secs; + } +} + +uint64_t PusherProxy::getRePublishCount(){ + return _republish_count; +} + } /* namespace mediakit */ diff --git a/src/Pusher/PusherProxy.h b/src/Pusher/PusherProxy.h index bee32d63..baeef029 100644 --- a/src/Pusher/PusherProxy.h +++ b/src/Pusher/PusherProxy.h @@ -43,6 +43,10 @@ public: */ void publish(const std::string& dstUrl) override; + int getStatus(); + uint64_t getLiveSecs(); + uint64_t getRePublishCount(); + private: // 重推逻辑函数 void rePublish(const std::string &dstUrl, int iFailedCnt); @@ -50,6 +54,11 @@ private: private: int _retry_count; toolkit::Timer::Ptr _timer; + toolkit::Ticker _live_ticker; + //0 表示正常 1 表示正在尝试推流 + std::atomic _live_status; + std::atomic _live_secs; + std::atomic _republish_count; std::weak_ptr _weak_src; std::function _on_close; std::function _on_publish;