Skip to content

Z
ZLMediaKit
Commit 7c89c165 by xiongguangjie
Options

add getProxyPusherInfo and getProxyInfo restful api

parent bfec1b1e
正在显示 包含 190 行增加 和 7 行删除
postman/ZLMediaKit.postman_collection.json
{
"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": [
......
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<recursive_mutex> 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<recursive_mutex> 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) {
......
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;
}
......@@ -59,9 +63,13 @@ void PlayerProxy::play(const string &strUrlTmp) {
if (!err) {
// 取消定时器,避免hls拉流索引文件因为网络波动失败重连成功后出现循环重试的情况
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 "<<strongSelf->_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 */
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<void(const toolkit::SockException &ex)> _on_close;
std::function<void(const toolkit::SockException &ex)> _on_play;
MultiMediaSourceMuxer::Ptr _muxer;
toolkit::Ticker _live_ticker;
//0 表示正常 1 表示正在尝试拉流
std::atomic<int> _live_status;
std::atomic<uint64_t> _live_secs;
std::atomic<uint64_t> _repull_count;
};
} /* namespace mediakit */
......
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 "<<strong_self->_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 */
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<int> _live_status;
std::atomic<uint64_t> _live_secs;
std::atomic<uint64_t> _republish_count;
std::weak_ptr<MediaSource> _weak_src;
std::function<void(const toolkit::SockException &ex)> _on_close;
std::function<void(const toolkit::SockException &ex)> _on_publish;
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论