amass/ZLMediaKit
1
0
Fork 0
mirror of https://github.com/ZLMediaKit/ZLMediaKit.git synced 2024-11-24 19:50:38 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

rtp server新增支持自定义vhost和app名称 (#3693)

Browse Source 
代码来自https://github.com/ZLMediaKit/ZLMediaKit/pull/3446 ， 增加了vhost
This commit is contained in:
zhang2349 2024-07-09 10:42:10 +08:00 committed by GitHub
parent b4fecdc929
commit c72e576420
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
14 changed files with 252 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
api/include/mk_rtp_server.h
View File

@ -24,6 +24,7 @@ typedef struct mk_rtp_server_t *mk_rtp_server;
* @return * @return
*/ */
API_EXPORT mk_rtp_server API_CALL mk_rtp_server_create(uint16_t port, int tcp_mode, const char *stream_id); API_EXPORT mk_rtp_server API_CALL mk_rtp_server_create(uint16_t port, int tcp_mode, const char *stream_id);
API_EXPORT mk_rtp_server API_CALL mk_rtp_server_create2(uint16_t port, int tcp_mode, const char *vhost, const char *app, const char *stream_id);
/** /**
* TCP 主动模式时连接到服务器是否成功的回调 * TCP 主动模式时连接到服务器是否成功的回调

13
api/source/mk_rtp_server.cpp
View File

@ -18,7 +18,13 @@ using namespace mediakit;
API_EXPORT mk_rtp_server API_CALL mk_rtp_server_create(uint16_t port, int tcp_mode, const char *stream_id) { API_EXPORT mk_rtp_server API_CALL mk_rtp_server_create(uint16_t port, int tcp_mode, const char *stream_id) {
RtpServer::Ptr *server = new RtpServer::Ptr(new RtpServer); RtpServer::Ptr *server = new RtpServer::Ptr(new RtpServer);
(*server)->start(port, stream_id, (RtpServer::TcpMode)tcp_mode); (*server)->start(port, MediaTuple { DEFAULT_VHOST, kRtpAppName, stream_id, "" }, (RtpServer::TcpMode)tcp_mode);
return (mk_rtp_server)server;
}
API_EXPORT mk_rtp_server API_CALL mk_rtp_server_create2(uint16_t port, int tcp_mode, const char *vhost, const char *app, const char *stream_id) {
RtpServer::Ptr *server = new RtpServer::Ptr(new RtpServer);
(*server)->start(port, MediaTuple { vhost, app, stream_id, "" }, (RtpServer::TcpMode)tcp_mode);
return (mk_rtp_server)server; return (mk_rtp_server)server;
} }
@ -71,6 +77,11 @@ API_EXPORT mk_rtp_server API_CALL mk_rtp_server_create(uint16_t port, int enable
return nullptr; return nullptr;
} }
API_EXPORT mk_rtp_server API_CALL mk_rtp_server_create2(uint16_t port, int tcp_mode, const char *vhost, const char *app, const char *stream_id) {
WarnL << "请打开ENABLE_RTPPROXY后再编译";
return nullptr;
}
API_EXPORT void API_CALL mk_rtp_server_release(mk_rtp_server ctx) { API_EXPORT void API_CALL mk_rtp_server_release(mk_rtp_server ctx) {
WarnL << "请打开ENABLE_RTPPROXY后再编译"; WarnL << "请打开ENABLE_RTPPROXY后再编译";
} }

80
postman/ZLMediaKit.postman_collection.json
View File

@ -1476,6 +1476,16 @@
"value": "{{ZLMediaKit_secret}}", "value": "{{ZLMediaKit_secret}}",
"description": "api操作密钥(配置文件配置)" "description": "api操作密钥(配置文件配置)"
}, },
{
"key": "vhost",
"value": "{{defaultVhost}}",
"description": "虚拟主机，例如__defaultVhost__"
},
{
"key": "app",
"value": "rtp",
"description": "应用名，例如 rtp"
},
{ {
"key": "stream_id", "key": "stream_id",
"value": "test", "value": "test",
@ -1517,6 +1527,16 @@
"value": "1", "value": "1",
"description": "tcp模式，0时为不启用tcp监听，1时为启用tcp监听，2时为tcp主动连接模式" "description": "tcp模式，0时为不启用tcp监听，1时为启用tcp监听，2时为tcp主动连接模式"
}, },
{
"key": "vhost",
"value": "{{defaultVhost}}",
"description": "虚拟主机，例如__defaultVhost__"
},
{
"key": "app",
"value": "rtp",
"description": "应用名，例如 rtp"
},
{ {
"key": "stream_id", "key": "stream_id",
"value": "test", "value": "test",
@ -1582,6 +1602,16 @@
"value": "1", "value": "1",
"description": "tcp模式，0时为不启用tcp监听，1时为启用tcp监听" "description": "tcp模式，0时为不启用tcp监听，1时为启用tcp监听"
}, },
{
"key": "vhost",
"value": "{{defaultVhost}}",
"description": "虚拟主机，例如__defaultVhost__"
},
{
"key": "app",
"value": "rtp",
"description": "应用名，例如 rtp"
},
{ {
"key": "stream_id", "key": "stream_id",
"value": "test", "value": "test",
@ -1635,6 +1665,16 @@
"value": "1", "value": "1",
"description": "tcp主动模式时服务端端口" "description": "tcp主动模式时服务端端口"
}, },
{
"key": "vhost",
"value": "{{defaultVhost}}",
"description": "虚拟主机，例如__defaultVhost__"
},
{
"key": "app",
"value": "rtp",
"description": "应用名，例如 rtp"
},
{ {
"key": "stream_id", "key": "stream_id",
"value": "test", "value": "test",
@ -1666,6 +1706,16 @@
"value": "{{ZLMediaKit_secret}}", "value": "{{ZLMediaKit_secret}}",
"description": "api操作密钥(配置文件配置)" "description": "api操作密钥(配置文件配置)"
}, },
{
"key": "vhost",
"value": "{{defaultVhost}}",
"description": "虚拟主机，例如__defaultVhost__"
},
{
"key": "app",
"value": "rtp",
"description": "应用名，例如 rtp"
},
{ {
"key": "stream_id", "key": "stream_id",
"value": "test", "value": "test",
@ -1697,6 +1747,16 @@
"value": "{{ZLMediaKit_secret}}", "value": "{{ZLMediaKit_secret}}",
"description": "api操作密钥(配置文件配置)" "description": "api操作密钥(配置文件配置)"
}, },
{
"key": "vhost",
"value": "{{defaultVhost}}",
"description": "虚拟主机，例如__defaultVhost__"
},
{
"key": "app",
"value": "rtp",
"description": "应用名，例如 rtp"
},
{ {
"key": "stream_id", "key": "stream_id",
"value": "test", "value": "test",
@ -1733,6 +1793,16 @@
"value": "{{ZLMediaKit_secret}}", "value": "{{ZLMediaKit_secret}}",
"description": "api操作密钥(配置文件配置)" "description": "api操作密钥(配置文件配置)"
}, },
{
"key": "vhost",
"value": "{{defaultVhost}}",
"description": "虚拟主机，例如__defaultVhost__"
},
{
"key": "app",
"value": "rtp",
"description": "应用名，例如 rtp"
},
{ {
"key": "stream_id", "key": "stream_id",
"value": "test", "value": "test",
@ -1764,6 +1834,16 @@
"value": "{{ZLMediaKit_secret}}", "value": "{{ZLMediaKit_secret}}",
"description": "api操作密钥(配置文件配置)" "description": "api操作密钥(配置文件配置)"
}, },
{
"key": "vhost",
"value": "{{defaultVhost}}",
"description": "虚拟主机，例如__defaultVhost__"
},
{
"key": "app",
"value": "rtp",
"description": "应用名，例如 rtp"
},
{ {
"key": "stream_id", "key": "stream_id",
"value": "test", "value": "test",

115
server/WebApi.cpp
View File

@ -477,18 +477,19 @@ Value makeMediaSourceJson(MediaSource &media){
} }
#if defined(ENABLE_RTPPROXY) #if defined(ENABLE_RTPPROXY)
uint16_t openRtpServer(uint16_t local_port, const string &stream_id, int tcp_mode, const string &local_ip, bool re_use_port, uint32_t ssrc, int only_track, bool multiplex) { uint16_t openRtpServer(uint16_t local_port, const mediakit::MediaTuple &tuple, int tcp_mode, const string &local_ip, bool re_use_port, uint32_t ssrc, int only_track, bool multiplex) {
if (s_rtp_server.find(stream_id)) { auto key = tuple.shortUrl();
//为了防止RtpProcess所有权限混乱的问题，不允许重复添加相同的stream_id if (s_rtp_server.find(key)) {
//为了防止RtpProcess所有权限混乱的问题，不允许重复添加相同的key
return 0; return 0;
} }
auto server = s_rtp_server.makeWithAction(stream_id, [&](RtpServer::Ptr server) { auto server = s_rtp_server.makeWithAction(key, [&](RtpServer::Ptr server) {
server->start(local_port, stream_id, (RtpServer::TcpMode)tcp_mode, local_ip.c_str(), re_use_port, ssrc, only_track, multiplex); server->start(local_port, tuple, (RtpServer::TcpMode)tcp_mode, local_ip.c_str(), re_use_port, ssrc, only_track, multiplex);
}); });
server->setOnDetach([stream_id](const SockException &ex) { server->setOnDetach([key](const SockException &ex) {
//设置rtp超时移除事件 //设置rtp超时移除事件
s_rtp_server.erase(stream_id); s_rtp_server.erase(key);
}); });
//回复json //回复json
@ -1199,7 +1200,15 @@ void installWebApi() {
api_regist("/index/api/getRtpInfo",[](API_ARGS_MAP){ api_regist("/index/api/getRtpInfo",[](API_ARGS_MAP){
CHECK_SECRET(); CHECK_SECRET();
CHECK_ARGS("stream_id"); CHECK_ARGS("stream_id");
auto src = MediaSource::find(DEFAULT_VHOST, kRtpAppName, allArgs["stream_id"]); std::string vhost = DEFAULT_VHOST;
if (!allArgs["vhost"].empty()) {
vhost = allArgs["vhost"];
}
std::string app = kRtpAppName;
if (!allArgs["app"].empty()) {
app = allArgs["app"];
}
auto src = MediaSource::find(vhost, app, allArgs["stream_id"]);
auto process = src ? src->getRtpProcess() : nullptr; auto process = src ? src->getRtpProcess() : nullptr;
if (!process) { if (!process) {
val["exist"] = false; val["exist"] = false;
@ -1212,7 +1221,16 @@ void installWebApi() {
api_regist("/index/api/openRtpServer",[](API_ARGS_MAP){ api_regist("/index/api/openRtpServer",[](API_ARGS_MAP){
CHECK_SECRET(); CHECK_SECRET();
CHECK_ARGS("port", "stream_id"); CHECK_ARGS("port", "stream_id");
std::string vhost = DEFAULT_VHOST;
if (!allArgs["vhost"].empty()) {
vhost = allArgs["vhost"];
}
std::string app = kRtpAppName;
if (!allArgs["app"].empty()) {
app = allArgs["app"];
}
auto stream_id = allArgs["stream_id"]; auto stream_id = allArgs["stream_id"];
auto tuple = MediaTuple { vhost, app, stream_id, "" };
auto tcp_mode = allArgs["tcp_mode"].as<int>(); auto tcp_mode = allArgs["tcp_mode"].as<int>();
if (allArgs["enable_tcp"].as<int>() && !tcp_mode) { if (allArgs["enable_tcp"].as<int>() && !tcp_mode) {
//兼容老版本请求，新版本去除enable_tcp参数并新增tcp_mode参数 //兼容老版本请求，新版本去除enable_tcp参数并新增tcp_mode参数
@ -1227,10 +1245,10 @@ void installWebApi() {
if (!allArgs["local_ip"].empty()) { if (!allArgs["local_ip"].empty()) {
local_ip = allArgs["local_ip"]; local_ip = allArgs["local_ip"];
} }
auto port = openRtpServer(allArgs["port"], stream_id, tcp_mode, local_ip, allArgs["re_use_port"].as<bool>(), auto port = openRtpServer(allArgs["port"], tuple, tcp_mode, local_ip, allArgs["re_use_port"].as<bool>(),
allArgs["ssrc"].as<uint32_t>(), only_track); allArgs["ssrc"].as<uint32_t>(), only_track);
if (port == 0) { if (port == 0) {
throw InvalidArgsException("该stream_id已存在"); throw InvalidArgsException("This stream already exists");
} }
//回复json //回复json
val["port"] = port; val["port"] = port;
@ -1239,7 +1257,16 @@ void installWebApi() {
api_regist("/index/api/openRtpServerMultiplex", [](API_ARGS_MAP) { api_regist("/index/api/openRtpServerMultiplex", [](API_ARGS_MAP) {
CHECK_SECRET(); CHECK_SECRET();
CHECK_ARGS("port", "stream_id"); CHECK_ARGS("port", "stream_id");
std::string vhost = DEFAULT_VHOST;
if (!allArgs["vhost"].empty()) {
vhost = allArgs["vhost"];
}
std::string app = kRtpAppName;
if (!allArgs["app"].empty()) {
app = allArgs["app"];
}
auto stream_id = allArgs["stream_id"]; auto stream_id = allArgs["stream_id"];
auto tuple = MediaTuple { vhost, app, stream_id, "" };
auto tcp_mode = allArgs["tcp_mode"].as<int>(); auto tcp_mode = allArgs["tcp_mode"].as<int>();
if (allArgs["enable_tcp"].as<int>() && !tcp_mode) { if (allArgs["enable_tcp"].as<int>() && !tcp_mode) {
// 兼容老版本请求，新版本去除enable_tcp参数并新增tcp_mode参数 // 兼容老版本请求，新版本去除enable_tcp参数并新增tcp_mode参数
@ -1254,9 +1281,10 @@ void installWebApi() {
if (!allArgs["local_ip"].empty()) { if (!allArgs["local_ip"].empty()) {
local_ip = allArgs["local_ip"]; local_ip = allArgs["local_ip"];
} }
auto port = openRtpServer(allArgs["port"], stream_id, tcp_mode, local_ip, true, 0, only_track,true);
auto port = openRtpServer(allArgs["port"], tuple, tcp_mode, local_ip, true, 0, only_track, true);
if (port == 0) { if (port == 0) {
throw InvalidArgsException("该stream_id已存在"); throw InvalidArgsException("This stream already exists");
} }
// 回复json // 回复json
val["port"] = port; val["port"] = port;
@ -1273,9 +1301,19 @@ void installWebApi() {
invoker(200, headerOut, val.toStyledString()); invoker(200, headerOut, val.toStyledString());
}; };
auto server = s_rtp_server.find(allArgs["stream_id"]); std::string vhost = DEFAULT_VHOST;
if (!allArgs["vhost"].empty()) {
vhost = allArgs["vhost"];
}
std::string app = kRtpAppName;
if (!allArgs["app"].empty()) {
app = allArgs["app"];
}
auto stream_id = allArgs["stream_id"];
auto tuple = MediaTuple { vhost, app, stream_id, "" };
auto server = s_rtp_server.find(tuple.shortUrl());
if (!server) { if (!server) {
cb(SockException(Err_other, "未找到rtp服务")); cb(SockException(Err_other, "can not find the stream"));
return; return;
} }
server->connectToServer(allArgs["dst_url"], allArgs["dst_port"], cb); server->connectToServer(allArgs["dst_url"], allArgs["dst_port"], cb);
@ -1285,7 +1323,17 @@ void installWebApi() {
CHECK_SECRET(); CHECK_SECRET();
CHECK_ARGS("stream_id"); CHECK_ARGS("stream_id");
if(s_rtp_server.erase(allArgs["stream_id"]) == 0){ std::string vhost = DEFAULT_VHOST;
if (!allArgs["vhost"].empty()) {
vhost = allArgs["vhost"];
}
std::string app = kRtpAppName;
if (!allArgs["app"].empty()) {
app = allArgs["app"];
}
auto stream_id = allArgs["stream_id"];
auto tuple = MediaTuple { vhost, app, stream_id, "" };
if (s_rtp_server.erase(tuple.shortUrl()) == 0) {
val["hit"] = 0; val["hit"] = 0;
return; return;
} }
@ -1296,7 +1344,17 @@ void installWebApi() {
CHECK_SECRET(); CHECK_SECRET();
CHECK_ARGS("stream_id", "ssrc"); CHECK_ARGS("stream_id", "ssrc");
auto server = s_rtp_server.find(allArgs["stream_id"]); std::string vhost = DEFAULT_VHOST;
if (!allArgs["vhost"].empty()) {
vhost = allArgs["vhost"];
}
std::string app = kRtpAppName;
if (!allArgs["app"].empty()) {
app = allArgs["app"];
}
auto stream_id = allArgs["stream_id"];
auto tuple = MediaTuple { vhost, app, stream_id, "" };
auto server = s_rtp_server.find(tuple.shortUrl());
if (!server) { if (!server) {
throw ApiRetException("RtpServer not found by stream_id", API::NotFound); throw ApiRetException("RtpServer not found by stream_id", API::NotFound);
} }
@ -1308,8 +1366,11 @@ void installWebApi() {
std::lock_guard<std::recursive_mutex> lck(s_rtp_server._mtx); std::lock_guard<std::recursive_mutex> lck(s_rtp_server._mtx);
for (auto &pr : s_rtp_server._map) { for (auto &pr : s_rtp_server._map) {
auto vec = split(pr.first, "/");
Value obj; Value obj;
obj["stream_id"] = pr.first; obj["vhost"] = vec[0];
obj["app"] = vec[1];
obj["stream_id"] = vec[2];
obj["port"] = pr.second->getPort(); obj["port"] = pr.second->getPort();
val["data"].append(obj); val["data"].append(obj);
} }
@ -1438,8 +1499,16 @@ void installWebApi() {
api_regist("/index/api/pauseRtpCheck", [](API_ARGS_MAP) { api_regist("/index/api/pauseRtpCheck", [](API_ARGS_MAP) {
CHECK_SECRET(); CHECK_SECRET();
CHECK_ARGS("stream_id"); CHECK_ARGS("stream_id");
std::string vhost = DEFAULT_VHOST;
if (!allArgs["vhost"].empty()) {
vhost = allArgs["vhost"];
}
std::string app = kRtpAppName;
if (!allArgs["app"].empty()) {
app = allArgs["app"];
}
//只是暂停流的检查，流媒体服务器做为流负载服务，收流就转发，RTSP/RTMP有自己暂停协议 //只是暂停流的检查，流媒体服务器做为流负载服务，收流就转发，RTSP/RTMP有自己暂停协议
auto src = MediaSource::find(DEFAULT_VHOST, kRtpAppName, allArgs["stream_id"]); auto src = MediaSource::find(vhost, app, allArgs["stream_id"]);
auto process = src ? src->getRtpProcess() : nullptr; auto process = src ? src->getRtpProcess() : nullptr;
if (process) { if (process) {
process->setStopCheckRtp(true); process->setStopCheckRtp(true);
@ -1451,7 +1520,15 @@ void installWebApi() {
api_regist("/index/api/resumeRtpCheck", [](API_ARGS_MAP) { api_regist("/index/api/resumeRtpCheck", [](API_ARGS_MAP) {
CHECK_SECRET(); CHECK_SECRET();
CHECK_ARGS("stream_id"); CHECK_ARGS("stream_id");
auto src = MediaSource::find(DEFAULT_VHOST, kRtpAppName, allArgs["stream_id"]); std::string vhost = DEFAULT_VHOST;
if (!allArgs["vhost"].empty()) {
vhost = allArgs["vhost"];
}
std::string app = kRtpAppName;
if (!allArgs["app"].empty()) {
app = allArgs["app"];
}
auto src = MediaSource::find(vhost, app, allArgs["stream_id"]);
auto process = src ? src->getRtpProcess() : nullptr; auto process = src ? src->getRtpProcess() : nullptr;
if (process) { if (process) {
process->setStopCheckRtp(false); process->setStopCheckRtp(false);

2
server/WebApi.h
View File

@ -202,7 +202,7 @@ void installWebApi();
void unInstallWebApi(); void unInstallWebApi();
#if defined(ENABLE_RTPPROXY) #if defined(ENABLE_RTPPROXY)
uint16_t openRtpServer(uint16_t local_port, const std::string &stream_id, int tcp_mode, const std::string &local_ip, bool re_use_port, uint32_t ssrc, int only_track, bool multiplex=false); uint16_t openRtpServer(uint16_t local_port, const mediakit::MediaTuple &tuple, int tcp_mode, const std::string &local_ip, bool re_use_port, uint32_t ssrc, int only_track, bool multiplex=false);
#endif #endif
Json::Value makeMediaSourceJson(mediakit::MediaSource &media); Json::Value makeMediaSourceJson(mediakit::MediaSource &media);

4
server/WebHook.cpp
View File

@ -682,7 +682,9 @@ void installWebHook() {
ArgsType body; ArgsType body;
body["local_port"] = local_port; body["local_port"] = local_port;
body["stream_id"] = stream_id; body[VHOST_KEY] = tuple.vhost;
body["app"] = tuple.app;
body["stream_id"] = tuple.stream;
body["tcp_mode"] = tcp_mode; body["tcp_mode"] = tcp_mode;
body["re_use_port"] = re_use_port; body["re_use_port"] = re_use_port;
body["ssrc"] = ssrc; body["ssrc"] = ssrc;

2
src/Common/config.h
View File

@ -107,7 +107,7 @@ extern const std::string kBroadcastReloadConfig;
// rtp server 超时 // rtp server 超时
extern const std::string kBroadcastRtpServerTimeout; extern const std::string kBroadcastRtpServerTimeout;
#define BroadcastRtpServerTimeoutArgs uint16_t &local_port, const string &stream_id,int &tcp_mode, bool &re_use_port, uint32_t &ssrc #define BroadcastRtpServerTimeoutArgs uint16_t &local_port, const MediaTuple &tuple, int &tcp_mode, bool &re_use_port, uint32_t &ssrc
// rtc transport sctp 连接状态 // rtc transport sctp 连接状态
extern const std::string kBroadcastRtcSctpConnecting; extern const std::string kBroadcastRtcSctpConnecting;

11
src/Rtp/RtpProcess.cpp
View File

@ -23,17 +23,14 @@ static constexpr size_t kMaxCachedFrameMS = 10 * 1000;
namespace mediakit { namespace mediakit {
RtpProcess::Ptr RtpProcess::createProcess(std::string stream_id) { RtpProcess::Ptr RtpProcess::createProcess(const MediaTuple &tuple) {
RtpProcess::Ptr ret(new RtpProcess(std::move(stream_id))); RtpProcess::Ptr ret(new RtpProcess(tuple));
ret->createTimer(); ret->createTimer();
return ret; return ret;
} }
RtpProcess::RtpProcess(string stream_id) { RtpProcess::RtpProcess(const MediaTuple &tuple) {
_media_info.schema = kRtpAppName; static_cast<MediaTuple &>(_media_info) = tuple;
_media_info.vhost = DEFAULT_VHOST;
_media_info.app = kRtpAppName;
_media_info.stream = std::move(stream_id);
GET_CONFIG(string, dump_dir, RtpProxy::kDumpDir); GET_CONFIG(string, dump_dir, RtpProxy::kDumpDir);
{ {

4
src/Rtp/RtpProcess.h
View File

@ -25,7 +25,7 @@ public:
using Ptr = std::shared_ptr<RtpProcess>; using Ptr = std::shared_ptr<RtpProcess>;
using onDetachCB = std::function<void(const toolkit::SockException &ex)>; using onDetachCB = std::function<void(const toolkit::SockException &ex)>;
static Ptr createProcess(std::string stream_id); static Ptr createProcess(const MediaTuple &tuple);
~RtpProcess(); ~RtpProcess();
enum OnlyTrack { kAll = 0, kOnlyAudio = 1, kOnlyVideo = 2 }; enum OnlyTrack { kAll = 0, kOnlyAudio = 1, kOnlyVideo = 2 };
@ -91,7 +91,7 @@ protected:
bool close(mediakit::MediaSource &sender) override; bool close(mediakit::MediaSource &sender) override;
private: private:
RtpProcess(std::string stream_id); RtpProcess(const MediaTuple &tuple);
void emitOnPublish(); void emitOnPublish();
void doCachedFunc(); void doCachedFunc();

24
src/Rtp/RtpServer.cpp
View File

@ -30,18 +30,18 @@ class RtcpHelper: public std::enable_shared_from_this<RtcpHelper> {
public: public:
using Ptr = std::shared_ptr<RtcpHelper>; using Ptr = std::shared_ptr<RtcpHelper>;
RtcpHelper(Socket::Ptr rtcp_sock, std::string stream_id) { RtcpHelper(Socket::Ptr rtcp_sock, MediaTuple tuple) {
_rtcp_sock = std::move(rtcp_sock); _rtcp_sock = std::move(rtcp_sock);
_stream_id = std::move(stream_id); _tuple = std::move(tuple);
} }
void setRtpServerInfo(uint16_t local_port, RtpServer::TcpMode mode, bool re_use_port, uint32_t ssrc, int only_track) { void setRtpServerInfo(uint16_t local_port, RtpServer::TcpMode mode, bool re_use_port, uint32_t ssrc, int only_track) {
_ssrc = ssrc; _ssrc = ssrc;
_process = RtpProcess::createProcess(_stream_id); _process = RtpProcess::createProcess(_tuple);
_process->setOnlyTrack((RtpProcess::OnlyTrack)only_track); _process->setOnlyTrack((RtpProcess::OnlyTrack)only_track);
_timeout_cb = [=]() mutable { _timeout_cb = [=]() mutable {
NOTICE_EMIT(BroadcastRtpServerTimeoutArgs, Broadcast::kBroadcastRtpServerTimeout, local_port, _stream_id, (int)mode, re_use_port, ssrc); NOTICE_EMIT(BroadcastRtpServerTimeoutArgs, Broadcast::kBroadcastRtpServerTimeout, local_port, _tuple, (int)mode, re_use_port, ssrc);
}; };
weak_ptr<RtcpHelper> weak_self = shared_from_this(); weak_ptr<RtcpHelper> weak_self = shared_from_this();
@ -117,12 +117,12 @@ private:
Ticker _ticker; Ticker _ticker;
Socket::Ptr _rtcp_sock; Socket::Ptr _rtcp_sock;
RtpProcess::Ptr _process; RtpProcess::Ptr _process;
std::string _stream_id; MediaTuple _tuple;
RtpProcess::onDetachCB _on_detach; RtpProcess::onDetachCB _on_detach;
std::shared_ptr<struct sockaddr_storage> _rtcp_addr; std::shared_ptr<struct sockaddr_storage> _rtcp_addr;
}; };
void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_mode, const char *local_ip, bool re_use_port, uint32_t ssrc, int only_track, bool multiplex) { void RtpServer::start(uint16_t local_port, const MediaTuple &tuple, TcpMode tcp_mode, const char *local_ip, bool re_use_port, uint32_t ssrc, int only_track, bool multiplex) {
//创建udp服务器 //创建udp服务器
auto poller = EventPollerPool::Instance().getPoller(); auto poller = EventPollerPool::Instance().getPoller();
Socket::Ptr rtp_socket = Socket::createSocket(poller, true); Socket::Ptr rtp_socket = Socket::createSocket(poller, true);
@ -148,9 +148,9 @@ void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_
UdpServer::Ptr udp_server; UdpServer::Ptr udp_server;
RtcpHelper::Ptr helper; RtcpHelper::Ptr helper;
//增加了多路复用判断，如果多路复用为true，就走else逻辑，同时保留了原来stream_id为空走else逻辑 //增加了多路复用判断，如果多路复用为true，就走else逻辑，同时保留了原来stream_id为空走else逻辑
if (!stream_id.empty() && !multiplex) { if (!tuple.stream.empty() && !multiplex) {
//指定了流id，那么一个端口一个流(不管是否包含多个ssrc的多个流，绑定rtp源后，会筛选掉ip端口不匹配的流) //指定了流id，那么一个端口一个流(不管是否包含多个ssrc的多个流，绑定rtp源后，会筛选掉ip端口不匹配的流)
helper = std::make_shared<RtcpHelper>(std::move(rtcp_socket), stream_id); helper = std::make_shared<RtcpHelper>(std::move(rtcp_socket), tuple);
helper->startRtcp(); helper->startRtcp();
helper->setRtpServerInfo(local_port, tcp_mode, re_use_port, ssrc, only_track); helper->setRtpServerInfo(local_port, tcp_mode, re_use_port, ssrc, only_track);
bool bind_peer_addr = false; bool bind_peer_addr = false;
@ -185,7 +185,9 @@ void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_
auto processor = helper ? helper->getProcess() : nullptr; auto processor = helper ? helper->getProcess() : nullptr;
// 如果共享同一个processor对象，那么tcp server深圳为单线程模式确保线程安全 // 如果共享同一个processor对象，那么tcp server深圳为单线程模式确保线程安全
tcp_server = std::make_shared<TcpServer>(processor ? poller : nullptr); tcp_server = std::make_shared<TcpServer>(processor ? poller : nullptr);
(*tcp_server)[RtpSession::kStreamID] = stream_id; (*tcp_server)[RtpSession::kVhost] = tuple.vhost;
(*tcp_server)[RtpSession::kApp] = tuple.app;
(*tcp_server)[RtpSession::kStreamID] = tuple.stream;
(*tcp_server)[RtpSession::kSSRC] = ssrc; (*tcp_server)[RtpSession::kSSRC] = ssrc;
(*tcp_server)[RtpSession::kOnlyTrack] = only_track; (*tcp_server)[RtpSession::kOnlyTrack] = only_track;
if (tcp_mode == PASSIVE) { if (tcp_mode == PASSIVE) {
@ -193,13 +195,13 @@ void RtpServer::start(uint16_t local_port, const string &stream_id, TcpMode tcp_
tcp_server->start<RtpSession>(local_port, local_ip, 1024, [weak_self, processor](std::shared_ptr<RtpSession> &session) { tcp_server->start<RtpSession>(local_port, local_ip, 1024, [weak_self, processor](std::shared_ptr<RtpSession> &session) {
session->setRtpProcess(processor); session->setRtpProcess(processor);
}); });
} else if (stream_id.empty()) { } else if (tuple.stream.empty()) {
// tcp主动模式时只能一个端口一个流，必须指定流id; 创建TcpServer对象也仅用于传参 // tcp主动模式时只能一个端口一个流，必须指定流id; 创建TcpServer对象也仅用于传参
throw std::runtime_error(StrPrinter << "tcp主动模式时必需指定流id"); throw std::runtime_error(StrPrinter << "tcp主动模式时必需指定流id");
} }
} }
_on_cleanup = [rtp_socket, stream_id]() { _on_cleanup = [rtp_socket]() {
if (rtp_socket) { if (rtp_socket) {
//去除循环引用 //去除循环引用
rtp_socket->setOnRead(nullptr); rtp_socket->setOnRead(nullptr);

2
src/Rtp/RtpServer.h
View File

@ -43,7 +43,7 @@ public:
* @param ssrc 指定的ssrc * @param ssrc 指定的ssrc
* @param multiplex 多路复用 * @param multiplex 多路复用
*/ */
void start(uint16_t local_port, const std::string &stream_id = "", TcpMode tcp_mode = PASSIVE, void start(uint16_t local_port, const MediaTuple &tuple = MediaTuple{DEFAULT_VHOST, kRtpAppName, "", ""}, TcpMode tcp_mode = PASSIVE,
const char *local_ip = "::", bool re_use_port = true, uint32_t ssrc = 0, int only_track = 0, bool multiplex = false); const char *local_ip = "::", bool re_use_port = true, uint32_t ssrc = 0, int only_track = 0, bool multiplex = false);
/** /**

14
src/Rtp/RtpSession.cpp
View File

@ -21,6 +21,8 @@ using namespace toolkit;
namespace mediakit{ namespace mediakit{
const string RtpSession::kVhost = "vhost";
const string RtpSession::kApp = "app";
const string RtpSession::kStreamID = "stream_id"; const string RtpSession::kStreamID = "stream_id";
const string RtpSession::kSSRC = "ssrc"; const string RtpSession::kSSRC = "ssrc";
const string RtpSession::kOnlyTrack = "only_track"; const string RtpSession::kOnlyTrack = "only_track";
@ -31,7 +33,9 @@ void RtpSession::attachServer(const Server &server) {
} }
void RtpSession::setParams(mINI &ini) { void RtpSession::setParams(mINI &ini) {
_stream_id = ini[kStreamID]; _tuple.vhost = ini[kVhost];
_tuple.app = ini[kApp];
_tuple.stream = ini[kStreamID];
_ssrc = ini[kSSRC]; _ssrc = ini[kSSRC];
_only_track = ini[kOnlyTrack]; _only_track = ini[kOnlyTrack];
int udp_socket_buffer = ini[kUdpRecvBuffer]; int udp_socket_buffer = ini[kUdpRecvBuffer];
@ -63,7 +67,7 @@ void RtpSession::onError(const SockException &err) {
if (_emit_detach) { if (_emit_detach) {
_process->onDetach(err); _process->onDetach(err);
} }
WarnP(this) << _stream_id << " " << err; WarnP(this) << _tuple.shortUrl() << " " << err;
} }
void RtpSession::onManager() { void RtpSession::onManager() {
@ -107,12 +111,12 @@ void RtpSession::onRtpPacket(const char *data, size_t len) {
} }
// 未指定流id就使用ssrc为流id // 未指定流id就使用ssrc为流id
if (_stream_id.empty()) { if (_tuple.stream.empty()) {
_stream_id = printSSRC(_ssrc); _tuple.stream = printSSRC(_ssrc);
} }
if (!_process) { if (!_process) {
_process = RtpProcess::createProcess(_stream_id); _process = RtpProcess::createProcess(_tuple);
_process->setOnlyTrack((RtpProcess::OnlyTrack)_only_track); _process->setOnlyTrack((RtpProcess::OnlyTrack)_only_track);
weak_ptr<RtpSession> weak_self = static_pointer_cast<RtpSession>(shared_from_this()); weak_ptr<RtpSession> weak_self = static_pointer_cast<RtpSession>(shared_from_this());
_process->setOnDetach([weak_self](const SockException &ex) { _process->setOnDetach([weak_self](const SockException &ex) {

4
src/Rtp/RtpSession.h
View File

@ -22,6 +22,8 @@ namespace mediakit{
class RtpSession : public toolkit::Session, public RtpSplitter { class RtpSession : public toolkit::Session, public RtpSplitter {
public: public:
static const std::string kVhost;
static const std::string kApp;
static const std::string kStreamID; static const std::string kStreamID;
static const std::string kSSRC; static const std::string kSSRC;
static const std::string kOnlyTrack; static const std::string kOnlyTrack;
@ -54,7 +56,7 @@ private:
int _only_track = 0; int _only_track = 0;
uint32_t _ssrc = 0; uint32_t _ssrc = 0;
toolkit::Ticker _ticker; toolkit::Ticker _ticker;
std::string _stream_id; MediaTuple _tuple;
struct sockaddr_storage _addr; struct sockaddr_storage _addr;
RtpProcess::Ptr _process; RtpProcess::Ptr _process;
}; };

2
tests/test_rtp.cpp
View File

@ -42,7 +42,7 @@ static bool loadFile(const char *path, const EventPoller::Ptr &poller) {
memset(&addr, 0, sizeof(addr)); memset(&addr, 0, sizeof(addr));
addr.ss_family = AF_INET; addr.ss_family = AF_INET;
auto sock = Socket::createSocket(poller); auto sock = Socket::createSocket(poller);
auto process = RtpProcess::createProcess("test"); auto process = RtpProcess::createProcess(MediaTuple { DEFAULT_VHOST, kRtpAppName, "test", "" });
uint64_t stamp_last = 0; uint64_t stamp_last = 0;
auto total_size = std::make_shared<size_t>(0); auto total_size = std::make_shared<size_t>(0);