amass/ZLMediaKit
1
0
Fork 0
mirror of https://github.com/ZLMediaKit/ZLMediaKit.git synced 2024-11-26 04:31:37 +08:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

调整代码细节

Browse Source
This commit is contained in:
ziyue 2021-06-17 10:12:34 +08:00
parent 06023f1770
commit 936c6f7965
1 changed files with 29 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

62
server/WebApi.cpp
View File

@ -232,15 +232,15 @@ static inline void addHttpListener(){
} }
//拉流代理器列表 //拉流代理器列表
static unordered_map<string ,PlayerProxy::Ptr> s_proxyMap; static unordered_map<string, PlayerProxy::Ptr> s_proxyMap;
static recursive_mutex s_proxyMapMtx; static recursive_mutex s_proxyMapMtx;
//推流代理器列表 //推流代理器列表
static unordered_map<string ,PusherProxy::Ptr> s_proxyPusherMap; static unordered_map<string, PusherProxy::Ptr> s_proxyPusherMap;
static recursive_mutex s_proxyPusherMapMtx; static recursive_mutex s_proxyPusherMapMtx;
//FFmpeg拉流代理器列表 //FFmpeg拉流代理器列表
static unordered_map<string ,FFmpegSource::Ptr> s_ffmpegMap; static unordered_map<string, FFmpegSource::Ptr> s_ffmpegMap;
static recursive_mutex s_ffmpegMapMtx; static recursive_mutex s_ffmpegMapMtx;
#if defined(ENABLE_RTPPROXY) #if defined(ENABLE_RTPPROXY)
@ -607,12 +607,12 @@ void installWebApi() {
const string &stream, const string &stream,
const string &url, const string &url,
int retryCount, int retryCount,
const function<void(const SockException &ex,const string &key)> &cb){ const function<void(const SockException &ex, const string &key)> &cb) {
auto key = getProxyKey(vhost, app, stream); auto key = getProxyKey(vhost, app, stream);
lock_guard<recursive_mutex> lck(s_proxyPusherMapMtx); lock_guard<recursive_mutex> lck(s_proxyPusherMapMtx);
if (s_proxyPusherMap.find(key) != s_proxyPusherMap.end()){ if (s_proxyPusherMap.find(key) != s_proxyPusherMap.end()) {
//已经在推流了 //已经在推流了
cb(SockException(Err_success),key); cb(SockException(Err_success), key);
return; return;
} }
@ -621,21 +621,21 @@ void installWebApi() {
if (retryCount != 0) retry_count = retryCount; if (retryCount != 0) retry_count = retryCount;
//添加推流代理 //添加推流代理
PusherProxy::Ptr pusher(new PusherProxy(schema,vhost, app, stream, retry_count, poller)); PusherProxy::Ptr pusher(new PusherProxy(schema, vhost, app, stream, retry_count, poller));
s_proxyPusherMap[key] = pusher; s_proxyPusherMap[key] = pusher;
//开始推流，如果推流失败或者推流中止，将会自动重试若干次，默认一直重试 //开始推流，如果推流失败或者推流中止，将会自动重试若干次，默认一直重试
pusher->setPushCallbackOnce([cb, key, url](const SockException &ex){ pusher->setPushCallbackOnce([cb, key, url](const SockException &ex) {
if (ex){ if (ex) {
InfoL << "key: " << key << ", " << "addStreamPusherProxy pusher callback error: " << ex.what(); InfoL << "key: " << key << ", " << "addStreamPusherProxy pusher callback error: " << ex.what();
lock_guard<recursive_mutex> lck(s_proxyMapMtx); lock_guard<recursive_mutex> lck(s_proxyMapMtx);
s_proxyMap.erase(key); s_proxyMap.erase(key);
} }
cb(ex,key); cb(ex, key);
}); });
//被主动关闭推流 //被主动关闭推流
pusher->setOnClose([key, url](const SockException &ex){ pusher->setOnClose([key, url](const SockException &ex) {
InfoL << "key: " << key << ", " << "addStreamPusherProxy close callback error: " << ex.what(); InfoL << "key: " << key << ", " << "addStreamPusherProxy close callback error: " << ex.what();
lock_guard<recursive_mutex> lck(s_proxyMapMtx); lock_guard<recursive_mutex> lck(s_proxyMapMtx);
s_proxyMap.erase(key); s_proxyMap.erase(key);
@ -647,50 +647,46 @@ void installWebApi() {
//测试url http://127.0.0.1/index/api/addStreamPusherProxy?schema=rtmp&vhost=__defaultVhost__&app=proxy&stream=0&dst_url=rtmp://127.0.0.1/live/obs //测试url http://127.0.0.1/index/api/addStreamPusherProxy?schema=rtmp&vhost=__defaultVhost__&app=proxy&stream=0&dst_url=rtmp://127.0.0.1/live/obs
api_regist("/index/api/addStreamPusherProxy", [](API_ARGS_MAP_ASYNC) { api_regist("/index/api/addStreamPusherProxy", [](API_ARGS_MAP_ASYNC) {
CHECK_SECRET(); CHECK_SECRET();
CHECK_ARGS("schema","vhost","app","stream"); CHECK_ARGS("schema", "vhost", "app", "stream", "dst_url");
InfoL << allArgs["schema"] << ", " << allArgs["vhost"] << ", " << allArgs["app"] << ", " << allArgs["stream"]; auto dst_url = allArgs["dst_url"];
auto src_url = allArgs["schema"] + "/" + allArgs["vhost"] + "/" + allArgs["app"] + "/" + allArgs["stream"];
//查找源
auto src = MediaSource::find(allArgs["schema"], auto src = MediaSource::find(allArgs["schema"],
allArgs["vhost"], allArgs["vhost"],
allArgs["app"], allArgs["app"],
allArgs["stream"]); allArgs["stream"]);
if (!src){ if (!src) {
InfoL << "addStreamPusherProxy， canont find source stream!"; WarnL << "addStreamPusherProxy, can not find source stream:" << src_url;
const_cast<Value &>(val)["code"] = API::OtherFailed; val["code"] = API::NotFound;
const_cast<Value &>(val)["msg"] = "can not find the source stream"; val["msg"] = "can not find the source stream";
invoker(200, headerOut, val.toStyledString()); invoker(200, headerOut, val.toStyledString());
return; return;
} }
std::string srcUrl = allArgs["schema"] + "://" + "127.0.0.1" + "/" + allArgs["app"] + "/" + allArgs["stream"]; InfoL << "addStreamPusherProxy, find stream: " << src_url << ", push dst url: " << dst_url;
std::string pushUrl = decodeBase64(allArgs["dst_url"]);
InfoL << "addStreamPusherProxy， find stream: " << srcUrl << ", push dst url: " << pushUrl;
addStreamPusherProxy(allArgs["schema"], addStreamPusherProxy(allArgs["schema"],
allArgs["vhost"], allArgs["vhost"],
allArgs["app"], allArgs["app"],
allArgs["stream"], allArgs["stream"],
pushUrl, allArgs["dst_url"],
allArgs["retry_count"], allArgs["retry_count"],
[invoker,val,headerOut, pushUrl](const SockException &ex, const string &key){ [invoker, val, headerOut, dst_url](const SockException &ex, const string &key) mutable {
if(ex){ if (ex) {
const_cast<Value &>(val)["code"] = API::OtherFailed; val["code"] = API::OtherFailed;
const_cast<Value &>(val)["msg"] = ex.what(); val["msg"] = ex.what();
InfoL << "Publish error url: " << pushUrl; WarnL << "Publish stream failed, dst url is: " << dst_url;
}else{ } else {
const_cast<Value &>(val)["data"]["key"] = key; val["data"]["key"] = key;
InfoL << "Publish success, Please play with player:" << pushUrl; InfoL << "Publish success, please play with player:" << dst_url;
} }
invoker(200, headerOut, val.toStyledString()); invoker(200, headerOut, val.toStyledString());
}); });
}); });
//关闭推流代理 //关闭推流代理
//测试url http://127.0.0.1/index/api/delStreamPusherProxy?key=__defaultVhost__/proxy/0 //测试url http://127.0.0.1/index/api/delStreamPusherProxy?key=__defaultVhost__/proxy/0
api_regist("/index/api/delStreamPusherProxy",[](API_ARGS_MAP){ api_regist("/index/api/delStreamPusherProxy", [](API_ARGS_MAP) {
CHECK_SECRET(); CHECK_SECRET();
CHECK_ARGS("key"); CHECK_ARGS("key");
lock_guard<recursive_mutex> lck(s_proxyPusherMapMtx); lock_guard<recursive_mutex> lck(s_proxyPusherMapMtx);